#include "util.h"
#include "hashdb.h"
#include "buffer_trans.h"
#include <sys/time.h>
static uint32 hash1(struct hdb *db,char *k,int len){
    uchar *key=(uchar*)k;
	uchar *it;
	uint32 h;
#define PRIME1		37
	for (it= key, h = 0; len--;)
		h = h * PRIME1 ^ (*it++ - ' ');
	h %= db->Bnum;
	return h;
}
struct hdb* hdbopen(char *filename,int flag,struct hparam *param){
    if(filename==NULL){
        fprintf(stderr,"must specify a database name\n");
		return NULL;
	}
	int fd,i;
	int dbflag=flag==0?O_RDWR|O_CREAT|O_TRUNC:flag;
	if((fd=open(filename,dbflag,DEF_FILE_MODE))<0){
	    fprintf(stderr,"can't open db file:%s\n",filename);
		return NULL;
	}
	/*char *mode;
	if(dbflag&O_RDONLY){
	    mode="rb";
	}else if(dbflag&O_RDWR){
	    mode="wb+";
	}else{
	    mode="wb+";
	}
	FILE *fp=fopen(filename,mode);
	if(fp==NULL){
	    perror("fopen");
		exit(1);
	}
	fd=fileno(fp);*/
	struct hdb *db=calloc(1,sizeof(struct hdb));
	if(db==NULL){
	    fprintf(stderr,"can't allocate memory\n");
		close(fd);
		return NULL;
	}
	if((db->dname=malloc(strlen(filename)+1))==NULL){
	    free(db);
		close(fd);
		return NULL;
	}
	memcpy(db->dname,filename,strlen(filename));
	db->dname[strlen(filename)]='\0';
	db->flag=dbflag;
	//db->fp=fp;
	int maplen;
	void *mapaddr;
	struct stat sbuf;
	fstat(fd,&sbuf);
	if(sbuf.st_size!=0){
    	DEBUG(printf("%s:file size=%d\n",__func__,sbuf.st_size));
		char buf[sizeof(struct hdbhead)];
		db->head=(struct hdbhead*)buf;
		readn(fd,buf,sizeof(buf));
		if(((db->Pgsiz&(db->Pgsiz-1))!=0)||db->Pgsiz<512||db->head->magic!=HASH_MAGIC){
		    printf("%s:bad hash db file\n",__func__);
			free(db->dname);
			free(db);
			close(fd);
			return NULL;
		}
		maplen=db->Bnum*sizeof(uint64)+BUCKETBEG;
		if((mapaddr=mmap(NULL,maplen,PROT_WRITE|PROT_READ,MAP_SHARED,fd,0))==MAP_FAILED){
	        free(db->dname);
			free(db);
			close(fd);
	        return NULL;
	    }
		db->head=(struct hdbhead*)mapaddr;
	}		
	else{
	    maplen=((param==NULL||param->bnum==0)?DEF_BUCKETS:param->bnum)*sizeof(uint64)+BUCKETBEG;
		int pgsiz=param==NULL||param->pgsiz==0?DEF_PAGESIZ:param->pgsiz;
		if(ftruncate(fd,(maplen+pgsiz-1)&~(pgsiz-1))<0) printf("%s:ftruncate error:%s\n",__func__,strerror(errno));
	    if((mapaddr=mmap(NULL,maplen,PROT_WRITE|PROT_READ,MAP_SHARED,fd,0))==MAP_FAILED){
	        free(db->dname);
			free(db);
			close(fd);
	        return NULL;
	    }
		db->head=(struct hdbhead*)mapaddr;
		db->head->magic=HASH_MAGIC;
		if(param==NULL){
		    db->Bnum=DEF_BUCKETS;
		    db->Align=DEF_ALIGN;
	        db->Pgsiz=DEF_PAGESIZ;
		    db->Minkey=DEF_MINKEY;
		}else{
		    db->Bnum=param->bnum==0?DEF_BUCKETS:param->bnum;
			db->Align=param->align==0?DEF_ALIGN:param->align;
			db->Pgsiz=param->pgsiz<512?DEF_PAGESIZ:param->pgsiz;
			db->Minkey=param->minkey==0?DEF_MINKEY:param->minkey;
		}
		db->Fstart=-1;
		db->Fnum=NFREEPG;
		db->Maxpg=(maplen+db->Pgsiz-1)&~(db->Pgsiz-1);
		db->Flast=0;
	}
	db->Fd=fd;
	db->oversiz=(db->Pgsiz-INDEXBEG)/(db->Minkey<<1)-sizeof(int16);
	if(param!=NULL){
	    db->cc=cachenew(fd,param->cache_nbucket,db->Pgsiz,param->cache_maxbytes,param->swapratio);
	}else{
	    db->cc=cachenew(fd,0,db->Pgsiz,0,0);
	}
	db->pstack=(struct pgstack*)calloc(db->Bnum,sizeof(struct pgstack));
	db->clock=calloc(db->Bnum,sizeof(int32));
	db->block=malloc(sizeof(pthread_rwlock_t)*db->Bnum);
	if(db->pstack==NULL||db->clock==NULL||db->block==NULL){
	    printf("%s:memory allocate error\n",__func__);
		exit(1);
	}
	for(i=0;i<db->Bnum;i++) pthread_rwlock_init(db->block+i,NULL);
	pthread_mutex_init(&db->dblock,NULL);
	db->buckets=mapaddr+BUCKETBEG;
	db->fpa=mapaddr+HEADSIZ;
	printf("%s:Fnum=%d oversiz=%d bnum=%d align=%d Maxpg=%d\n",__func__,db->Fnum,db->oversiz,db->Bnum,db->Align,db->Maxpg);
	return db;
}
uint64 getfpg(struct hdb *db){
    uint64 pgno;
	pthread_mutex_lock(&db->dblock);
	if(db->Nfpg>0){
	    if(db->Fstart>=0){
	        while(db->Fstart!=db->Fend){
		        pgno=db->fpa[db->Fstart];
				if(db->Fend<0) db->Fend=db->Fstart;
			    db->Fstart=(db->Fstart+1)%db->Fnum;
				db->Nfpg--;
			    if(pgno<db->Maxpg){
				    pthread_mutex_unlock(&db->dblock);
					if(db->flag&HDB_IN_TRANSACTION) STACK_PUSH(db->newpgno,uint64,pgno);
					return pgno;
				}
		    }
			if(db->Nfpg==0){
				pgno=db->Maxpg;
				db->Maxpg+=db->Pgsiz;
				db->Flast=0;
				db->Fstart=-1;
				db->Nfpg=0;
				pthread_mutex_unlock(&db->dblock);
				if(db->flag&HDB_IN_TRANSACTION) STACK_PUSH(db->newpgno,uint64,pgno);
				return pgno;
			}
			//pgno=db->fpa[(db->Fstart+db->Fnum-1)%db->Fnum];
			db->Fend=db->Fnum;
			if(db->Fread!=0) pgno=db->Fread;
			else{
			    pgno=db->Flast;
				db->Flast=0;
			}
			while(pgno!=0){
			    while(pgno!=0&&db->Fend>0){
				    struct freepage fp;
					if(pgno<db->Maxpg)
					    db->fpa[--db->Fend]=pgno;
			        seekread(db->Fd,(char*)&fp,FREEHEAD,pgno);
				    pgno=fp.nextpg;
				    //printf("%s:get a new pgno=%d\n",__func__,pgno);
			    }
				if(db->Fend==0||db->Flast==0) break;
				pgno=db->Flast;
				db->Flast=0;
			}
		    if(db->Fend==db->Fnum){
				db->Fstart=-1;
				db->Flast=0;
				printf("%s:error:nfpg=%d\n",__func__,db->Nfpg);
				exit(1);
			}
			else{
			    db->Fread=pgno;
				pgno=db->fpa[db->Fend];
				db->Nfpg--;
				db->Fstart=db->Fend+1;
				db->Fend=0;
			}
		}else{
		    printf("%s:error in get free page:db->Nfpg=%d\n",__func__,db->Nfpg);
		    struct freepage fp;
			pgno=db->Flast;
			seekread(db->Fd,(char*)&fp,FREEHEAD,pgno);
			db->Flast=fp.nextpg;
		}
	}
	else{
	    pgno=db->Maxpg;
		db->Maxpg+=db->Pgsiz;
	}
	//printf("%s:flast=%d pgno=%d db->Maxpg=%d\n",__func__,db->Flast,pgno,db->Maxpg);
	pthread_mutex_unlock(&db->dblock);
	if(db->flag&HDB_IN_TRANSACTION) STACK_PUSH(db->newpgno,uint64,pgno);
	return pgno;
}
int freepg(struct hdb *db,uint64 pgno){
    struct freepage fp={FREEPAGEMAGIC};
	if(db->flag&HDB_IN_TRANSACTION){
	    STACK_PUSH(db->delpgno,uint64,pgno);
		return true;
	}
	pthread_mutex_lock(&db->dblock);
	DEBUG(printf("%s:free pgno=%d Maxpg=%d fstart=%d fend=%d flast=%d fread=%d nfpg=%d\n",__func__,pgno,db->Maxpg,db->Fstart,db->Fend,db->Flast,db->Fread,db->Nfpg));
	if(pgno+db->Pgsiz==db->Maxpg){
	    db->Maxpg-=db->Pgsiz;
		pthread_mutex_unlock(&db->dblock);
		return true;
	}
	if(db->Fstart<0||db->Nfpg==0){
		db->fpa[0]=pgno;
		db->Fstart=0;
		db->Fend=1;
		//db->Flast=pgno;
	}else if(db->Fend>0&&db->Fend!=db->Fstart){
		db->fpa[db->Fend]=pgno;
		db->Fend=(db->Fend+1)%db->Fnum;
		if(db->Fend==db->Fstart) db->Fend=-1;
		//db->Flast=pgno;
	}else{
	    fp.nextpg=db->Flast;
	    db->Flast=pgno;
		DEBUG(printf("%s:pgno=%d fp->nextpg=%d\n",__func__,pgno,fp.nextpg));
	    seekwrite(db->Fd,(char*)&fp,sizeof(struct freepage),pgno);
	}/*
	if(db->Fstart>=0){
	    int i;
		for(i=0;i<db->Fnum;i++){
		    printf("%s:freepage[%d]=%d\n",__func__,i,db->fpa[i]);
		}
	}else printf("%s:no freepage\n",__func__);*/
	db->Nfpg++;
	pthread_mutex_unlock(&db->dblock);
	return true;
}
int getoverflowpage(struct hdb *db,uint64 pgno,int offset,char *buf,int ln){
    int gotsiz=0;
	char pbuf[BLOCKSIZ];
	char *ptr=buf;
	struct overflowpage *p=NULL;
	int start,cpsiz,rdsiz,presiz=0;
	int osiz=db->Pgsiz-OFLOWHEAD;
	p=(struct overflowpage*)pbuf;
	if(ln<=0) return 0;
	//printf("%s:pgno=%d offset=%d ln=%d osiz=%d\n",__func__,pgno,offset,ln,osiz);
	while(presiz<offset-osiz&&pgno!=0){//locate the page
	    seekread(db->Fd,pbuf,OFLOWHEAD,pgno);
		pgno=p->nextpg;
		presiz+=osiz;
	}
	if(offset+ln<=osiz+presiz)
	    return seekread(db->Fd,buf,ln,pgno+offset-presiz+OFLOWHEAD);
	if(db->Pgsiz>BLOCKSIZ) p=malloc(db->Pgsiz);
	
	start=offset-presiz;
	//printf("%s:presiz=%d start=%d pgno=%d\n",__func__,presiz,start,pgno);
	while(pgno!=0&&gotsiz<ln){
	    rdsiz=ln-gotsiz>db->Pgsiz-OFLOWHEAD?db->Pgsiz:ln-gotsiz+OFLOWHEAD;
		if(seekread(db->Fd,(char*)p,rdsiz,pgno)!=rdsiz){
		    if((char*)p!=pbuf) free(p);
			//printf("%s:read fail\n",__func__);
			return gotsiz;
		}
		cpsiz=osiz-start>ln-gotsiz?ln-gotsiz:osiz-start;
		memcpy(ptr,pbuf+OFLOWHEAD+start,cpsiz);
		ptr+=cpsiz;
		gotsiz+=cpsiz;
		//printf("%s:pgno=%d rdsiz=%d cpsiz=%d\n",__func__,pgno,rdsiz,cpsiz);
		start=0;
		pgno=p->nextpg;
	}
	if((char*)p!=pbuf) free(p);
	return gotsiz;
}
uint64 writeoverflowpage(struct hdb *db,char *key,int ksiz,char *val,int vsiz){   
	char *ptr=NULL,*buff;
	int siz=0,cpsiz,osiz=db->Pgsiz-OFLOWHEAD;
	int tot=ksiz+vsiz;
	uint64 pgno,nextpgno,retpgno;
	char buf[BLOCKSIZ];
	if(tot<=0) return true;
	nextpgno=getfpg(db);
	retpgno=nextpgno;
	if(db->Pgsiz>BLOCKSIZ) buff=malloc(db->Pgsiz);
	else buff=buf;
	struct overflowpage *p=(struct overflowpage*)buff;
	while(siz<tot){
	    ptr=buff+OFLOWHEAD;
		p->siz=0;
		//printf("%s:tot=%d siz=%d ksiz=%d vsiz=%d\n",__func__,tot,siz,ksiz,vsiz);
		pgno=nextpgno;
		if(tot-siz<=osiz){
		    cpsiz=ksiz-siz;
		    if(cpsiz>0){
			    //printf("%s:copy key %d bytes\n",__func__,ksiz-siz);
 			    memcpy(ptr,key+siz,cpsiz);
			    ptr+=cpsiz;
				siz+=cpsiz;
				p->siz+=cpsiz;
			}
			if(siz<tot){
			    cpsiz=tot-siz;
			    memcpy(ptr,val+(siz-ksiz),cpsiz);
				ptr+=cpsiz;
				p->siz+=cpsiz;
				siz=tot;
				//printf("%s:copy val %d bytes siz-ksiz=%d\n",__func__,ptr-buf,siz-ksiz);
			}
			p->nextpg=0;
		}
		else{
	        if(siz<ksiz){
		        cpsiz=ksiz-siz>osiz?osiz:ksiz-siz;
			    memcpy(ptr,key+siz,cpsiz);
			    siz+=cpsiz;
			    ptr+=cpsiz;
				p->siz+=cpsiz;
		    }
		    //printf("%s:write %d bytes\n",__func__,cpsiz);
		    if(ptr-buff<db->Pgsiz&&siz<tot){
		        cpsiz=tot-siz>db->Pgsiz-(ptr-buff)?db->Pgsiz-(ptr-buff):tot-siz;
		        memcpy(ptr,val+siz-ksiz,cpsiz);
		        siz+=cpsiz;
		        ptr+=cpsiz;
				p->siz+=cpsiz;
		    }
			nextpgno=p->nextpg=getfpg(db);
		}
        //printf("%s:siz=%d tot=%d ptr-buf=%d\n",__func__,siz,tot,ptr-buf);					
		//dbwrite(db,buff,ptr-buff,pgno);
		seekwrite(db->Fd,buff,db->Pgsiz,pgno);
		DEBUG(printf("%s:write %d bytes pgno=%d\n",__func__,ptr-buff,pgno));
		//printf("%s:retpgno=%d\n",__func__,retpgno);
	}
	//printf("%s:ptr-buff=%d write siz=%d\n",__func__,ptr-buff,siz);
	if(buff!=buf)free(buff);
	return retpgno;
}
int deloverflowpage(struct hdb *db,uint64 pgno,int offset,int siz){
    if(pgno==0) return true;
	char buf[OFLOWHEAD];
	struct overflowpage *p=(struct overflowpage*)buf;
	while(pgno!=0&&offset>db->Pgsiz-OFLOWHEAD){
	    seekread(db->Fd,buf,OFLOWHEAD,pgno);
		//freepg(db,pgno);
		pgno=p->nextpg;
		offset-=p->siz;
	}
	if(pgno==0) return false;
	seekread(db->Fd,buf,OFLOWHEAD,pgno);
	if(offset!=0){
	    pgno=p->nextpg;
	    p->nextpg=0;
	    seekwrite(db->Fd,buf,OFLOWHEAD,pgno);
	}else{
	    freepg(db,pgno);
		pgno=p->nextpg;
	}
	while(pgno!=0&&siz>0){
		seekread(db->Fd,buf,OFLOWHEAD,pgno);
		siz-=p->siz;
		freepg(db,pgno);
		pgno=p->nextpg;
	}
	return true;
}

struct idx* createindex(struct hdb *db,char *key,int ksiz,char *val,int vsiz,uint64 pgno,char *ibuf,int *siz){
    struct idx *index;
	int ssiz=*siz;
	if(ksiz==0) return NULL;
	if(IDXSIZ(ksiz,vsiz)<=db->oversiz){
	    *siz=IDXSIZ(ksiz,vsiz);
		if(ssiz<*siz||ibuf==NULL)
	        index=malloc(*siz);
		else  index=(struct idx*)ibuf;
		memcpy(index->d,key,ksiz);
		memcpy(index->d+ksiz,val,vsiz);
		WRITEIDX(index,ksiz,vsiz,PADSIZ(ksiz+vsiz),pgno);
	}
	else if(IDXKSIZ(ksiz)<=db->oversiz){
	    *siz=IDXKSIZ(ksiz);
		if(ssiz<*siz||ibuf==NULL)
	        index=malloc(*siz);
		else  index=(struct idx*)ibuf;
		memcpy(index->d+sizeof(uint64),key,ksiz);
		*((uint64*)index->d)=writeoverflowpage(db,NULL,0,val,vsiz);
		WRITEIDX(index,ksiz,vsiz,V_OFLOW,pgno);
	}
	else{
	    *siz=db->oversiz;
		if(ssiz<*siz||ibuf==NULL)
	        index=malloc(*siz);
		else  index=(struct idx*)ibuf;
		int wrsiz=db->oversiz-sizeof(uint64)-SIZEOFIDX;
		memcpy(index->d+sizeof(uint64),key,wrsiz);
		*((uint64*)index->d)=writeoverflowpage(db,key+wrsiz,ksiz-wrsiz,val,vsiz);
		WRITEIDX(index,ksiz,vsiz,PK_OFLOW,pgno);
	}
	return index;
}
struct rcd *createrec(struct hdb *db,char *key,int ksiz,char *val,int vsiz,char *rbuf,int *siz){
    struct rcd *rec=(struct rcd*)rbuf;
	int rsiz;
	if(RECSIZ(ksiz,vsiz)<=RECOVERSIZ){
	    rsiz=RECSIZ(ksiz,vsiz);
		if(*siz<rsiz) rec=malloc(rsiz);
		*siz=rsiz;
		//printf("%s:PADSIZ(%d+%d)=%d recsiz=%d\n",__func__,ksiz,vsiz,PADSIZ(ksiz+vsiz),rsiz);
		WRITEREC(rec,ksiz,vsiz,PADSIZ(ksiz+vsiz));
		memcpy(rec->d,key,ksiz);
		memcpy(rec->d+ksiz,val,vsiz);
	}else if(RECKSIZ(ksiz)<=RECOVERSIZ){
	    rsiz=RECKSIZ(ksiz);
		if(*siz<rsiz) rec=malloc(rsiz);
		*siz=rsiz;
		WRITEREC(rec,ksiz,vsiz,V_OFLOW);
		memcpy(rec->d+sizeof(uint64),key,ksiz);
		*((uint64*)rec->d)=writeoverflowpage(db,NULL,0,val,vsiz);
	}else{
	    rsiz=RECOVERSIZ;
		if(*siz<rsiz) rec=malloc(rsiz);
		*siz=rsiz;
		WRITEREC(rec,ksiz,vsiz,PK_OFLOW);
		int wrsiz=rsiz-sizeof(uint64)-SIZEOFREC;
		memcpy(rec->d+sizeof(uint64),key,wrsiz);
		DEBUG(printf("%s:PK_OFLOW:write %d bytes\n",__func__,wrsiz));
		*((uint64*)rec->d)=writeoverflowpage(db,key+wrsiz,ksiz-wrsiz,val,vsiz);
	}
	return rec;
}
int splitnode(struct hdb *db,struct page *p,uint16 ipos,struct idx *widx,int wsiz,struct idx **ridx,int *rsiz){
	struct page *bp,*np;
	char buf[BLOCKSIZ<<1];
	uint16 inum=(p->lower-INDEXBEG)/sizeof(uint16);
	int isplit,tsiz,i;
	struct idx *tidx;
	uint64 pgno;
	int32 siz=0;
	int half=(db->Pgsiz-INDEXBEG)>>1;
	if(db->Pgsiz>BLOCKSIZ){
		bp=malloc(db->Pgsiz<<1);
		np=(struct page*)((char*)bp+db->Pgsiz);
	}
	else{
		bp=(struct page*)buf;
		np=(struct page*)(buf+BLOCKSIZ);
	}
	for(isplit=0;isplit<inum;isplit++){
	    tidx=(struct idx*)((char*)p+p->index[isplit]);
		siz+=padidx(db,tidx)+sizeof(int16);
		DEBUG(printf("%s:i=%d idxsiz=%d half=%d\n",__func__,isplit,padidx(db,tidx),half));
		if(siz>=half){
			if(siz>half&&ipos<isplit&&isplit>=db->Minkey)
			    isplit--;
			else if(siz==half&&ipos+1>isplit&&isplit<=db->Minkey)
			    isplit++;
		    break;
		}
	}
	np->lower=INDEXBEG;
	np->upper=db->Pgsiz;
	pgno=getfpg(db);
	int start=(isplit==ipos&&isplit>=db->Minkey)?isplit:isplit+1;
	np->upper=db->Pgsiz;
	DEBUG(printf("%s:ipos=%d isplit=%d siz=%d start=%d n rpgno=%d\n",__func__,ipos,isplit,siz,start,pgno));
	for(i=start;i<inum;i++){
	    tidx=(struct idx*)((char*)p+p->index[i]);
		tsiz=padidx(db,tidx);
		np->upper-=tsiz;
		memcpy((char*)np+np->upper,tidx,tsiz);
		np->index[i-start]=np->upper;
	}
	np->lower=INDEXBEG+sizeof(int16)*(inum-start);
	bp->upper=db->Pgsiz;
	for(i=0;i<start;i++){
	    tidx=(struct idx*)((char*)p+p->index[i]);
		tsiz=padidx(db,tidx);
		bp->upper-=tsiz;
		memcpy((char*)bp+bp->upper,tidx,tsiz);
		bp->index[i]=bp->upper;
	}
	p->lower=INDEXBEG+start*sizeof(int16);
	memcpy(p->index,bp->index,p->lower-INDEXBEG);
	p->upper=bp->upper;
	memcpy((char*)p+p->upper,(char*)bp+bp->upper,db->Pgsiz-bp->upper);
	if(ipos==isplit&&isplit>=db->Minkey){
	    if(*rsiz<wsiz) *ridx=malloc(wsiz);
		*rsiz=wsiz;
		memcpy(*ridx,widx,wsiz);
		np->left=widx->pgno;
		DEBUG(printf("%s:split self left inum=%d right inum=%d\n",__func__,(p->lower-INDEXBEG)/sizeof(int16),(np->lower-INDEXBEG)/sizeof(int16)));
	}else{
	    tidx=(struct idx*)(bp->index[isplit]+(char*)bp);
		tsiz=padidx(db,tidx);
		if(tsiz>*rsiz) *ridx=malloc(tsiz);
		*rsiz=tsiz;
		memcpy(*ridx,tidx,tsiz);
		np->left=tidx->pgno;
		p->upper+=tsiz;
		p->lower-=sizeof(int16);
		DEBUG(printf("%s:left inum=%d\n",__func__,(p->lower-INDEXBEG)/sizeof(int16)));
	    if(ipos>isplit){
	        int nipos=ipos-start;
	        np->upper-=wsiz;
		    memmove(np->index+nipos+1,np->index+nipos,(inum-start-nipos)*sizeof(int16));
		    np->index[nipos]=np->upper;
		    memcpy(np->upper+(char*)np,widx,wsiz);
		    DEBUG(printf("%s:!!right:idx->offset=%d nipos=%d isiz=%d\n",__func__,widx->pgno,nipos,wsiz));
		    np->lower+=sizeof(int16);
	    }
	    else{
	        p->upper-=wsiz;
		    memmove(p->index+ipos+1,p->index+ipos,(start-ipos)*sizeof(int16));
		    p->index[ipos]=p->upper;
		    memcpy((char*)p+p->upper,widx,wsiz);
		    p->lower+=sizeof(int16);
			DEBUG(printf("%s:!!left remsiz=%d\n",__func__,p->upper-p->lower));
	    }
	}
	(*ridx)->pgno=pgno;
	addpage(db->cc,pgno,(char*)np,F_DIRTY);
	if((char*)bp!=buf){
		free(bp);
	}
	return true;
}
int splitleaf(struct hdb *db,struct leaf *lf,struct rcd *rec,int rsiz,struct idx **nidx,int *nsiz,uint32 hv){
    struct pgstack *pstack=db->pstack+hv;
	struct pcursor *cursor=pstack->stack+pstack->pnum-1;
	int ipos=cursor->idxoff;
	char buf[BLOCKSIZ<<1];
	struct leaf *blf,*rlf;
	uint64 pgno;
	int inum=(lf->lower-LEAFIDXBEG)/sizeof(uint16);
	int i,isplit,half=(db->Pgsiz-LEAFIDXBEG)>>1,siz=0,tsiz;
	struct rcd *it;
	if(db->Pgsiz>BLOCKSIZ){
	    blf=malloc(db->Pgsiz<<1);
		rlf=(struct leaf*)((char*)blf+db->Pgsiz);
	}
	else{
	    blf=(struct leaf*)buf;
		rlf=(struct leaf*)(buf+BLOCKSIZ);
	}
	for(isplit=0;isplit<inum;isplit++){
	    it=(struct rcd*)((char*)lf+lf->index[isplit]);
		DEBUG(printf("%s:recsiz=%d value=%s\n",__func__,sizeofrec(db,it),it->d));
		siz+=sizeofrec(db,it)+sizeof(uint16);
		if(siz>=half){
		    if(siz>half&&ipos<isplit&&isplit>=db->Minkey)
			    isplit--;
		    break;
		}
	}
	rlf->upper=db->Pgsiz;
	pgno=getfpg(db);
	int start=(isplit==ipos&&isplit>=db->Minkey)?isplit:isplit+1;
	DEBUG(printf("%s:ipos=%d isplit=%d siz=%d start=%d new rleaf=%d\n",__func__,ipos,isplit,siz,start,pgno));
	for(i=start;i<inum;i++){
	    it=(struct rcd*)((char*)lf+lf->index[i]);
		tsiz=sizeofrec(db,it);
		rlf->upper-=tsiz;
		memcpy((char*)rlf+rlf->upper,it,tsiz);
		rlf->index[i-start]=rlf->upper;
	}
	rlf->lower=LEAFIDXBEG+sizeof(int16)*(inum-start);
	blf->upper=db->Pgsiz;
	for(i=0;i<start;i++){
	    it=(struct rcd*)((char*)lf+lf->index[i]);
		tsiz=sizeofrec(db,it);
		blf->upper-=tsiz;
		memcpy((char*)blf+blf->upper,it,tsiz);
		blf->index[i]=blf->upper;
	}
	lf->lower=LEAFIDXBEG+start*sizeof(int16);
	memcpy(lf->index,blf->index,lf->lower-LEAFIDXBEG);
	lf->upper=blf->upper;
	memcpy((char*)lf+lf->upper,(char*)blf+blf->upper,db->Pgsiz-blf->upper);
	if(ipos==isplit&&isplit>=db->Minkey){
	    if(*nsiz<rsiz+sizeof(uint64)) *nidx=malloc(rsiz+sizeof(uint64));
		*nsiz=rsiz+sizeof(uint64);
		RECTOIDX(rec,rsiz,LEAFPGNO(pgno),*nidx);
		DEBUG(printf("%s:split self left inum=%d right inum=%d\n",__func__,(lf->lower-LEAFIDXBEG)/sizeof(int16),(rlf->lower-LEAFIDXBEG)/sizeof(int16)));
	}else{
	    it=(struct rcd*)(blf->index[isplit]+(char*)blf);
		tsiz=sizeofrec(db,it);
		if(tsiz+sizeof(uint64)>*nsiz) *nidx=malloc(tsiz+sizeof(uint64));
		*nsiz=tsiz+sizeof(uint64);
		RECTOIDX(it,tsiz,LEAFPGNO(pgno),*nidx);
		lf->upper+=tsiz;
		lf->lower-=sizeof(int16);
		DEBUG(printf("%s:left inum=%d\n",__func__,(lf->lower-LEAFIDXBEG)/sizeof(int16)));
	    if(ipos>isplit){
	        int nipos=ipos-start;
	        rlf->upper-=rsiz;
		    memmove(rlf->index+nipos+1,rlf->index+nipos,(inum-start-nipos)*sizeof(int16));
		    rlf->index[nipos]=rlf->upper;
		    memcpy(rlf->upper+(char*)rlf,rec,rsiz);
		    DEBUG(printf("%s:!!right:nipos=%d rsiz=%d\n",__func__,nipos,rsiz));
		    rlf->lower+=sizeof(int16);
	    }
	    else{
	        lf->upper-=rsiz;
		    memmove(lf->index+ipos+1,lf->index+ipos,(start-ipos)*sizeof(int16));
		    lf->index[ipos]=lf->upper;
		    memcpy((char*)lf+lf->upper,rec,rsiz);
		    lf->lower+=sizeof(int16);
			DEBUG(printf("%s:!!left remsiz=%d\n",__func__,lf->upper-lf->lower));
	    }
	}
	addpage(db->cc,pgno,(char*)rlf,F_DIRTY);
	if((char*)blf!=buf){
		free(blf);
	}
	return true;
}
int putidx(struct hdb *db,struct page *p,int pi,struct idx *index,int isiz,uint32 hv){
	struct pgstack *pstack=db->pstack+hv;
	struct pcursor *cursor=pstack->stack+pi;
	int16 ipos=cursor->idxoff;
	uint16 inum=(p->lower-INDEXBEG)/sizeof(int16);
	char buf[2][BLOCKSIZ];
	int i=0;
	int nsiz=BLOCKSIZ;
	struct idx *nidx=(struct idx*)buf[i];
	uint64 pgno;
	DEBUG(printf("%s:inum=%d ipos=%d pi=%d remsiz=%d isiz=%d\n",__func__,inum,ipos,pi,p->upper-p->lower,isiz));
	if(p->upper-p->lower>=isiz+sizeof(uint16)){
	    p->upper-=isiz;
		memcpy((char*)p+p->upper,index,isiz);
		memmove(p->index+ipos+1,p->index+ipos,(inum-ipos)*sizeof(int16));
		p->lower+=sizeof(int16);
		p->index[ipos]=p->upper;
		putpage(db->cc,(char*)p,F_DIRTY);
	}else{
	    splitnode(db,p,ipos,index,isiz,&nidx,&nsiz);
		if(pi==0){
		    pgno=getfpg(db);
			DEBUG(printf("%s:create left pgno=%d\n",__func__,pgno));
			addpage(db->cc,pgno,(char*)p,F_DIRTY);
			p->left=pgno;
			p->lower=INDEXBEG+sizeof(int16);
			p->upper=db->Pgsiz-nsiz;
			p->index[0]=p->upper;
			memcpy((char*)p+p->upper,nidx,nsiz);
			putpage(db->cc,(char*)p,F_DIRTY);
			return true;
		}
		putpage(db->cc,(char*)p,F_DIRTY);
		//addpage(db->cc,npgno,(char*)np,F_DIRTY);
		pi--;//don't use stackpop,maybe insert into a page different from stack top
		while(pi>=1&&nidx!=NULL){
			struct page *sp=(struct page*)getpage(db->cc,pstack->stack[pi].pgno,F_WRITE);
			DEBUG(printf("%s:innner pgno=%d remsiz=%d nsiz=%d\n",__func__,pstack->stack[pi].pgno,sp->upper-sp->lower,nsiz));
			ipos=pstack->stack[pi].idxoff;
			if(sp->upper-sp->lower>=nsiz+sizeof(uint16)){
			    inum=(sp->lower+(char*)sp-(char*)&sp->index[0])/sizeof(uint16);
				sp->upper-=nsiz;
				memcpy((char*)sp+sp->upper,nidx,nsiz);
				memmove(sp->index+ipos+1,sp->index+ipos,sizeof(uint16)*(inum-ipos));
				sp->index[ipos]=sp->upper;
				sp->lower+=sizeof(uint16);
				if((char*)nidx!=buf[i]) free(nidx);
				putpage(db->cc,(char*)sp,F_DIRTY);
				//if((char*)np!=pbuf) free(np);
				return true;
			}
			else{
				inum=(sp->lower-INDEXBEG)/sizeof(uint16);
				if(inum<db->Minkey&&pi!=0){
					printf("error:node has %d index,cannot split\n",inum);
			        return false;//mem leak!
				}
				int nextsiz=BLOCKSIZ;
				struct idx *nextidx=(struct idx*)buf[!i];
				ipos=pstack->stack[pi].idxoff;
				splitnode(db,sp,ipos,nidx,nsiz,&nextidx,&nextsiz);
				if((char*)nidx!=buf[i]) free(nidx);
				nidx=nextidx;
				nsiz=nextsiz;
				i=!i;
				//addpage(db->cc,npgno,(char*)np,F_DIRTY);
			}
			putpage(db->cc,(char*)sp,F_DIRTY);
			pi--;
		}
		if(nidx!=NULL){//pi=0
			struct page *sp=(struct page*)getpage(db->cc,pstack->stack[0].pgno,F_WRITE);
		    ipos=pstack->stack[0].idxoff;
			DEBUG(printf("%s:root page nsiz=%d remsiz=%d ipos=%d pgno=%d\n",__func__,nsiz,sp->upper-sp->lower,ipos,pstack->stack[0].pgno));
			if(sp->upper-sp->lower>=nsiz+sizeof(uint16)){
			    inum=(sp->lower+(char*)sp-(char*)&sp->index[0])/sizeof(uint16);
				sp->upper-=nsiz;
				memcpy((char*)sp+sp->upper,nidx,nsiz);
				memmove(sp->index+ipos+1,sp->index+ipos,sizeof(uint16)*(inum-ipos));
				sp->index[ipos]=sp->upper;
				sp->lower+=sizeof(uint16);
				if((char*)nidx!=buf[i]) free(nidx);
			}
			else{
				int inum=(sp->lower-INDEXBEG)/sizeof(uint16);
				if(inum<db->Minkey){
					printf("error:node has %d index,cannot split\n",inum);
					exit(1);
			        return false;//mem leak!
				}
				struct idx *nextidx=(struct idx*)buf[!i];
				int nextsiz=BLOCKSIZ;
				ipos=pstack->stack[pi].idxoff;
				splitnode(db,sp,ipos,nidx,nsiz,&nextidx,&nextsiz);
				uint64 leftpg=getfpg(db);
				DEBUG(printf("%s:root:get new leftpgno=%d\n",__func__,pgno));
				//np->parent=sp->parent=pstack->stack[pi].pgno;
				addpage(db->cc,leftpg,(char*)sp,F_DIRTY);
				//addpage(db->cc,npgno,(char*)np,F_DIRTY);
				sp->left=leftpg;
				sp->lower=INDEXBEG+sizeof(uint16);
				sp->upper=db->Pgsiz-nextsiz;
                //sp->parent=0;				
				sp->index[0]=sp->upper;
				memcpy((char*)sp+sp->upper,(char*)nextidx,nextsiz);
				if((char*)nidx!=buf[i])free(nidx);
				if((char*)nextidx!=buf[!i]) free(nextidx);		
			}
			putpage(db->cc,(char*)sp,F_DIRTY);
		}
	}
	//if((char*)np!=pbuf) free(np);
	return true;
}
bool putrec(struct hdb *db,struct leaf *lf,struct rcd *rec,int rsiz,uint32 hv){
    struct pgstack *pstack=db->pstack+hv;
	struct pcursor *cursor=pstack->stack+pstack->pnum-1;
	int16 ipos=cursor->idxoff;
	uint16 inum=(lf->lower-LEAFIDXBEG)/sizeof(uint16);
	char buf[BLOCKSIZ];
	//printf("%s:inum=%d stacknum=%d ipos=%d recsiz=%d remsiz=%d\n",__func__,inum,pstack->pnum,ipos,rsiz,lf->upper-lf->lower);//getchar();
	if(lf->upper-lf->lower>=rsiz+sizeof(uint16)){
	    lf->upper-=rsiz;
		memcpy((char*)lf+lf->upper,rec,rsiz);
		memmove(lf->index+ipos+1,lf->index+ipos,(inum-ipos)*sizeof(int16));
		lf->lower+=sizeof(int16);
		lf->index[ipos]=lf->upper;
		putpage(db->cc,(char*)lf,F_DIRTY);
		return true;
	}else{
	    struct idx *nidx=(struct idx*)buf;
		int nsiz=BLOCKSIZ;
		uint64 pgno;
		struct page *p;
		//struct leaf *rlf=(struct leaf*)lbuf;
		//if(db->Pgsiz>BLOCKSIZ) rlf=malloc(db->Pgsiz);
	    splitleaf(db,lf,rec,rsiz,&nidx,&nsiz,hv);
		//npgno=getfpg(db);
		//nidx->pgno=LEAFPGNO(npgno);
		if(pstack->pnum==1){
		    pgno=getfpg(db);
		    //dbwrite(db,(char*)lf,db->Pgsiz,pgno);
			addpage(db->cc,pgno,(char*)lf,F_DIRTY);
			DEBUG(printf("%s:create a leftleaf pgno=%d inum=%d rootpgno=%d nsiz=%d\n",__func__,pgno,(lf->lower-LEAFIDXBEG)/sizeof(int16),PGNO(pstack->stack[0].pgno),nsiz));
			struct page *p=(struct page*)lf;
			p->left=LEAFPGNO(pgno);
			p->lower=INDEXBEG+sizeof(int16);
			p->upper=db->Pgsiz-nsiz;
			p->index[0]=p->upper;
			memcpy((char*)p+p->upper,nidx,nsiz);
			//addpage(db->cc,PGNO(nidx->pgno),(char*)rlf,F_DIRTY);
			//addpage(db->cc,pgno,(char*)p,F_DIRTY);
			pgno=PGNO(pstack->stack[0].pgno);
			putpage(db->cc,(char*)lf,F_DIRTY);
			db->buckets[hv]=pgno;
		}else{
		    //addpage(db->cc,nidx->pgno,(char*)rlf,F_DIRTY);
			putpage(db->cc,(char*)lf,F_DIRTY);
		    p=(struct page*)getpage(db->cc,pstack->stack[pstack->pnum-2].pgno,F_WRITE);
		    putidx(db,p,pstack->pnum-2,nidx,nsiz,hv);
		}
		if((char*)nidx!=buf) free(nidx);
		//if((char*)rlf!=lbuf) free(rlf);
	}
	return true;
}
void* locate(struct hdb *db,char *key,uint32 ksiz,uint32 hv,int *hit){
    char buf[BLOCKSIZ],*sbuf=NULL,*tbuf=NULL;
	struct page *p=NULL;
	struct leaf *lp=NULL;
	struct idx *index;
	struct rcd *rec;
	int left,right,mid,inum;
	int cmp,slen,len;
	struct pgstack *pstack=db->pstack+hv;
	if(pstack->stack==NULL){
		pstack->stack=calloc(DEF_LEVEL,sizeof(struct pcursor));
		if(pstack->stack==NULL){
		    printf("%s:memory alloc error\n",__func__);
			exit(1);
		}
		#ifdef __DB_DEBUG
		else{
		    printf("%s:hv=%d pstack->stack=%d\n",__func__,hv,pstack->stack);
		}
		#endif
		pstack->pmax=DEF_LEVEL;
	}
	uint64 pgno=db->buckets[hv];
	pstack->pnum=0;
	*hit=0;
	//struct timeval beg,end;
	//gettimeofday(&beg,NULL);
	while(!ISLEAF(pgno)){
	    //if(pgno==0) exit(1);
	    p=(struct page*)getpage(db->cc,pgno,F_READ);
		inum=(p->lower-INDEXBEG)/sizeof(uint16);
		left=0;
		right=inum-1;
		DEBUG(printf("%s:internal page:%d inum=%d\n",__func__,pgno,inum));
		//db->locatenum++;
		while(left<=right){
		    mid=(left+right)>>1;
		    index=(struct idx*)(p->index[mid]+(char*)p);
			char *nkey;
		    uint32 flag=index->fap&I_FLAGMASK&~I_DELETED;
			//printf("%s:mid=%d fap=%x flag=%x\n",__func__,mid,index->fap,flag);
		    if(flag==0||flag==V_OFLOW){
		        len=index->ksiz>ksiz?ksiz:index->ksiz;
			    nkey=flag==V_OFLOW?index->d+sizeof(uint64):index->d;
		        cmp=strncmp(key,nkey,len);
			    if(cmp==0) cmp=ksiz-index->ksiz;
		    }
		    else if(flag==PK_OFLOW){
		        len=db->oversiz-sizeof(uint64)-SIZEOFIDX>ksiz?ksiz:db->oversiz-sizeof(uint64)-SIZEOFIDX;
			    nkey=index->d+sizeof(uint64);
				//printf("%s:nkey=%s\n",__func__,nkey);
			    if((cmp=strncmp(key,nkey,len))==0){
			        if(ksiz>len){
					    if(index->ksiz-len>BLOCKSIZ){
						    if(sbuf==NULL){
						        sbuf=(char*)malloc(index->ksiz-len);
							    slen=index->ksiz-len;
						    }else if(slen<index->ksiz-len){
							    sbuf=(char*)realloc(sbuf,index->ksiz-len);
								slen=index->ksiz-len;
							}
							tbuf=sbuf;
						}else tbuf=buf;
				        getoverflowpage(db,*((uint64*)index->d),0,tbuf,index->ksiz-len);
					    int ln=len;
					    len=index->ksiz>ksiz?ksiz:index->ksiz;
					    len-=ln;
						//printf("%s:tbuf=%s\n",__func__,tbuf);
					    cmp=strncmp(key+ln,tbuf,len);
					    if(cmp==0) cmp=ksiz-index->ksiz;
				    }
				    else cmp=-1;
			    }
				//printf("%s:cmp=%d\n",__func__,cmp);
			}
			else{printf("%s:error in locate inner key fap=%x\n",__func__,index->fap);exit(1);}
			//printf("%s:mid=%d nkey=%s\n",__func__,mid,nkey);
			if(cmp<0) right=mid-1;
            else if(cmp>0) left=mid+1;
            else{//find the same key
				STACKPUSH(db->pstack+hv,pgno,mid);
				*hit=1;
			    if(tbuf!=buf&&tbuf!=NULL) free(tbuf);
			    return p;
            }
		}
		STACKPUSH(db->pstack+hv,pgno,left);
		DEBUG(printf("%s:ipos=%d\n",__func__,left));
		if(left==0)
		    pgno=p->left;
		else
		    pgno=((struct idx*)(p->index[left-1]+(char*)p))->pgno;
		putpage(db->cc,(char*)p,0);
	}
	//gettimeofday(&end,NULL);
	//db->plocatetime+=end.tv_sec-beg.tv_sec+(end.tv_usec-beg.tv_usec)/1000000.0;
	if(ISLEAF(pgno)){
	    //struct timeval b,e;
		//gettimeofday(&b,NULL);
		lp=(struct leaf*)getpage(db->cc,PGNO(pgno),F_READ);
		//gettimeofday(&e,NULL);
		//db->cmptime+=e.tv_sec-b.tv_sec+(e.tv_usec-b.tv_usec)/1000000.0;
		inum=(lp->lower-LEAFIDXBEG)/sizeof(int16);
		left=0;
		right=inum-1;
		//db->locatenum++;
		
		//printf("%s:leaf inum=%d page:%d\n",__func__,inum,pgno);
		
		while(left<=right){
		    //db->cmpnum++;
		    mid=(left+right)>>1;
		    rec=(struct rcd*)(lp->index[mid]+(char*)lp);
			char *nkey;
		    uint32 flag=rec->fap&I_FLAGMASK;
			DEBUG(printf("%s:mid=%d fap=%x\n",__func__,mid,rec->fap));
		    if(flag==0||flag==V_OFLOW){
		        len=rec->ksiz>ksiz?ksiz:rec->ksiz;
			    nkey=(flag==V_OFLOW)?rec->d+sizeof(uint64):rec->d;
		        cmp=strncmp(key,nkey,len);
			    if(cmp==0) cmp=ksiz-rec->ksiz;
		    }
		    else if(flag==PK_OFLOW){
		        len=RECOVERSIZ-SIZEOFREC-sizeof(uint64)>ksiz?ksiz:RECOVERSIZ-sizeof(uint64)-SIZEOFREC;
			    nkey=rec->d+sizeof(uint64);
				//printf("%s:nkey=%s\n",__func__,nkey);
			    if((cmp=strncmp(key,nkey,len))==0){
			        if(ksiz>len){
					    if(rec->ksiz-len>BLOCKSIZ){
						    if(sbuf==NULL){
						        sbuf=malloc(rec->ksiz-len);
							    slen=rec->ksiz-len;
							}else if(slen<rec->ksiz-len){
							    sbuf=realloc(sbuf,rec->ksiz-len);
								slen=rec->ksiz-len;
							}
							tbuf=sbuf;
						}else tbuf=buf;
				        getoverflowpage(db,*((uint64*)rec->d),0,tbuf,rec->ksiz-len);
						//printf("%s:over key=%s\n",__func__,tbuf);
					    int ln=len;
					    len=rec->ksiz>ksiz?ksiz:rec->ksiz;
					    len-=ln;
					    cmp=strncmp(key+ln,tbuf,len);
					    if(cmp==0) cmp=ksiz-rec->ksiz;
				    }
				    else cmp=-1;
			    }
			}
			else{printf("%s:error in locate leaf key fap=%x pgno=%d\n",__func__,rec->fap,pgno);exit(1);}
			//gettimeofday(&e,NULL);
			//db->cmptime+=e.tv_sec-b.tv_sec+(e.tv_usec-b.tv_usec)/1000000.0;
			//gettimeofday(&b,NULL);
			if(cmp<0) right=mid-1;
            else if(cmp>0) left=mid+1;
            else{//find the same key
				STACKPUSH(db->pstack+hv,pgno,mid);
				*hit=1;
			    if(tbuf!=buf&&tbuf!=NULL) free(tbuf);
			    return lp;
            }
			
		}
		STACKPUSH(db->pstack+hv,pgno,left);
		
	}else printf("%s:error!last page is not leaf\n",__func__);
	//gettimeofday(&beg,NULL);
	//db->llocatetime+=beg.tv_sec-end.tv_sec+(beg.tv_usec-end.tv_usec)/1000000.0;
	if(tbuf!=buf&&tbuf!=NULL) free(tbuf);
	return lp;
}
bool hdbput(struct hdb *db,char *key,int ksiz,char *val,int vsiz,int flag){
    uint32 it=hash1(db,key,ksiz);
	if(pthread_rwlock_wrlock(db->block+it)!=0){
	    printf("%s:lock fail\n",__func__);
		exit(1);
	}
	uint64 pgno=db->buckets[it];
	struct idx *index;
	struct rcd *rec;
	char buf[BLOCKSIZ];
	char rbuf[BLOCKSIZ];
	int hit,rsiz=BLOCKSIZ;
	struct page *p;
	struct leaf *lp;
	void *op;
	uint64 ipgno=0;
	//struct timeval beg,end;
	db->clock[it]++;
	DEBUG(printf("\n%s:--------------hash value=%d\n",__func__,it));
	if(pgno==0){
	    if(db->Pgsiz>BLOCKSIZ) lp=(struct leaf*)malloc(db->Pgsiz);
	    else lp=(struct leaf*)buf;
	    rec=createrec(db,key,ksiz,val,vsiz,rbuf,&rsiz);
		pgno=getfpg(db);
		//printf("%s:bucket is null getfreepage=%d recsiz=%d rec->fap=0x%x\n",__func__,pgno,rsiz,rec->fap);
		lp->upper=db->Pgsiz-rsiz;
		lp->lower=LEAFIDXBEG+sizeof(int16);
		lp->index[0]=lp->upper;
		memcpy((char*)lp+lp->upper,rec,rsiz);
		db->buckets[it]=LEAFPGNO(pgno);
		addpage(db->cc,pgno,(char*)lp,F_DIRTY);
		if((char*)lp!=buf) free(lp);
		if(rbuf!=(char*)rec) free(rec);
		pthread_rwlock_unlock(db->block+it);
		pthread_mutex_lock(&db->dblock);
		db->Nrec++;//
		pthread_mutex_unlock(&db->dblock);		
		return true;
	}
	//gettimeofday(&beg,NULL);
	op=locate(db,key,ksiz,it,&hit);
	//gettimeofday(&end,NULL);
	//db->locatetime+=end.tv_sec-beg.tv_sec+(end.tv_usec-beg.tv_usec)/1000000.0;
	if(hit){
	    DEBUG(printf("%s:find a same key:%s pnum=%d\n",__func__,key,db->pstack[it].pnum));
	    struct pcursor *cursor=&(db->pstack[it].stack[db->pstack[it].pnum-1]);
		if(!ISLEAF(cursor->pgno)){
		    DEBUG(printf("%s:same key in innerpage idxoff=%d\n",__func__,cursor->idxoff));
		    p=op;
		    index=(struct idx*)((char*)p+p->index[cursor->idxoff]);
		    if(flag==PUT_REPLACE||(index->fap&I_DELETED)){
			    DEBUG(printf("%s:going to replace idx index->fap=0x%x\n",__func__,index->fap));
		        int isiz=BLOCKSIZ;
				ipgno=index->pgno;
			    delindex(db,p,cursor->idxoff);
				index=createindex(db,key,ksiz,val,vsiz,ipgno,buf,&isiz);
				DEBUG(printf("%s:idxoff=%d inum=%d\n",__func__,cursor->idxoff,(p->lower-INDEXBEG)/sizeof(int16)));
				putidx(db,p,db->pstack[it].pnum-1,index,isiz,it);
				if((char*)index!=buf) free(index);
				pthread_rwlock_unlock(db->block+it);
				return true;
		    }else if(flag==PUT_APPEND){
		        appendval(db,p,key,ksiz,val,vsiz,it);
				pthread_rwlock_unlock(db->block+it);
				return true;
		    }else{
			    pthread_rwlock_unlock(db->block+it);
		        return false;
		    }
		}else{
		    lp=op;
			#ifdef __DB_DEBUG
			if((lp->lower-LEAFIDXBEG)/sizeof(int16)>16){
			    int x;
				for(x=0;x<16;x++){
				    struct rcd *lrec=(struct rcd*)((char*)lp+lp->index[x]);
					printf("%s:%d rec->d=%s\n",__func__,x,lrec->d);
				}
			}
			#endif
		    rec=(struct rcd*)((char*)lp+lp->index[cursor->idxoff]);
			DEBUG(printf("%s:leaf duplicate:idxoff=%d rec->d=%s\n",__func__,cursor->idxoff,rec->d));
			if(flag==PUT_REPLACE||flag==0){
			    delrecord(db,lp,cursor->idxoff);
				rec=createrec(db,key,ksiz,val,vsiz,rbuf,&rsiz);
				putrec(db,lp,rec,rsiz,it);
				if((char*)rec!=rbuf) free(rec);
			}else if(flag==PUT_APPEND){
			    appendrec(db,lp,key,ksiz,val,vsiz,it);
			}else{
			    pthread_rwlock_unlock(db->block+it);
			    return false;
			}
			pthread_rwlock_unlock(db->block+it);
			return true;
		}
	}
	lp=op;
	//gettimeofday(&beg,NULL);
	rec=createrec(db,key,ksiz,val,vsiz,rbuf,&rsiz);
	putrec(db,lp,rec,rsiz,it);
	//gettimeofday(&end,NULL);
	//db->puttime+=end.tv_sec-beg.tv_sec+(end.tv_usec-beg.tv_usec)/1000000.0;
	
	if((char*)rec!=rbuf) free(rec);
	//check(db->cc);
	pthread_rwlock_unlock(db->block+it);
	pthread_mutex_lock(&db->dblock);
	db->Nrec++;
	pthread_mutex_unlock(&db->dblock);
	return true;
}
int appendval(struct hdb *db,struct page *p,char *key,int ksiz,char *val,int vsiz,uint32 hv){
    struct pcursor *cursor=&(db->pstack[hv].stack[db->pstack[hv].pnum-1]);
	struct idx *index=(struct idx*)((char*)p+p->index[cursor->idxoff]);
	uint64 pgno;
	int siz;
	char buf[BLOCKSIZ];
	struct overflowpage *op=(struct overflowpage*)buf;
	DEBUG(printf("%s:append vsiz=%d\n",__func__,vsiz));
	if(index->fap&V_OFLOW){
	    pgno=*((uint64*)index->d);
		siz=index->vsiz;
		index->vsiz+=vsiz;
		while(siz>db->Pgsiz-OFLOWHEAD){
			seekread(db->Fd,buf,OFLOWHEAD,pgno);
			siz-=db->Pgsiz-OFLOWHEAD;
			pgno=op->nextpg;
		}
		if(BLOCKSIZ<db->Pgsiz) op=malloc(db->Pgsiz);
		seekread(db->Fd,(char*)p,db->Pgsiz,pgno);
		int wrsiz=vsiz<db->Pgsiz-siz-OFLOWHEAD?vsiz:db->Pgsiz-siz-OFLOWHEAD;
		char *ptr=val;
		memcpy((char*)op+OFLOWHEAD+siz,ptr,wrsiz);
		vsiz-=wrsiz;
		ptr+=wrsiz;
		op->siz=wrsiz+siz;
		while(vsiz>0){
			op->nextpg=getfpg(db);
			seekwrite(db->Fd,(char*)op,db->Pgsiz,pgno);
			wrsiz=vsiz>db->Pgsiz-OFLOWHEAD?db->Pgsiz-OFLOWHEAD:vsiz;
			memcpy((char*)p+OFLOWHEAD,ptr,wrsiz);
			ptr+=wrsiz;
			pgno=op->nextpg;
			op->siz=wrsiz;
			vsiz-=wrsiz;
		}
		op->nextpg=0;
		seekwrite(db->Fd,(char*)op,db->Pgsiz,pgno);
		putpage(db->cc,(char*)p,F_DIRTY);
		if((char*)op!=buf) free(op);
	}else if(index->fap&PK_OFLOW){
	    uint64 pgno=*((uint64*)index->d);
		int siz=index->vsiz+index->ksiz-(db->oversiz-SIZEOFIDX-sizeof(uint64));
		op=(struct overflowpage*)buf;
		while(pgno!=0&&siz>db->Pgsiz-OFLOWHEAD){
			seekread(db->Fd,buf,OFLOWHEAD,pgno);
			siz-=db->Pgsiz-OFLOWHEAD;
			pgno=op->nextpg;
		}
		index->vsiz+=vsiz;
		if(BLOCKSIZ<db->Pgsiz) op=malloc(db->Pgsiz);
		seekread(db->Fd,(char*)op,db->Pgsiz,pgno);
		int wrsiz=vsiz<db->Pgsiz-siz-OFLOWHEAD?vsiz:db->Pgsiz-siz-OFLOWHEAD;
		char *ptr=val;
		memcpy((char*)op+OFLOWHEAD+siz,ptr,wrsiz);
		vsiz-=wrsiz;
		ptr+=wrsiz;
		op->siz=wrsiz+siz;
		while(vsiz>0){
			op->nextpg=getfpg(db);
			//op->siz=db->Pgsiz-OFLOWHEAD;
			DEBUG(printf("%s:append index pgno=%d bytes=%d\n",__func__,pgno,op->siz));
			seekwrite(db->Fd,(char*)op,db->Pgsiz,pgno);
			wrsiz=vsiz>db->Pgsiz-OFLOWHEAD?db->Pgsiz-OFLOWHEAD:vsiz;
			memcpy((char*)op+OFLOWHEAD,ptr,wrsiz);
			ptr+=wrsiz;
			pgno=op->nextpg;
			op->siz=wrsiz;
			vsiz-=wrsiz;
		}
		op->nextpg=0;
		seekwrite(db->Fd,(char*)op,db->Pgsiz,pgno);
		putpage(db->cc,(char*)p,F_DIRTY);
		if((char*)op!=buf) free(op);
	}
	else{
	    uint64 ipgno=index->pgno;
		int ilen=BLOCKSIZ,nvsiz;
		char vbuf[BLOCKSIZ],*nval;
		nval=vbuf;
		if(index->vsiz+vsiz>BLOCKSIZ) nval=(char*)malloc(index->vsiz+vsiz);
		getidxval(db,index,nval,index->vsiz);
		nvsiz=index->vsiz;
		memcpy(nval+nvsiz,val,vsiz);
		nvsiz+=vsiz;
		delindex(db,p,cursor->idxoff);
		index=createindex(db,key,ksiz,nval,nvsiz,ipgno,buf,&ilen);
		putidx(db,p,db->pstack[hv].pnum-1,index,ilen,hv);
		if((char*)index!=buf) free(index);
		if(nval!=vbuf) free(nval);
	}
	return true;
}
int getidxval(struct hdb *db,struct idx *index,char *val,int siz){
    uint32 flag=index->fap&I_FLAGMASK;
	int cpsiz=siz>index->vsiz?index->vsiz:siz;
	if(flag==0) memcpy(val,index->d+index->ksiz,cpsiz);
	else if(flag==V_OFLOW) getoverflowpage(db,*((uint64*)index->d),0,val,cpsiz);
	else if(flag==PK_OFLOW) getoverflowpage(db,*((uint64*)index->d),index->ksiz-(db->oversiz-SIZEOFIDX-sizeof(uint64)),val,cpsiz);
	else return false;
	return true;
}
int getrecval(struct hdb *db,struct rcd *rec,char *val,int siz){
    uint32 flag=rec->fap&I_FLAGMASK;
	int cpsiz=siz>rec->vsiz?rec->vsiz:siz;
	if(flag==0) memcpy(val,rec->d+rec->ksiz,cpsiz);
	else if(flag==V_OFLOW) getoverflowpage(db,*((uint64*)rec->d),0,val,cpsiz);
	else if(flag==PK_OFLOW) getoverflowpage(db,*((uint64*)rec->d),rec->ksiz-(RECOVERSIZ-SIZEOFREC-sizeof(uint64)),val,cpsiz);
	else return false;
	return true;
}
int appendrec(struct hdb* db,struct leaf *lf,char *key,int ksiz,char *val,int vsiz,uint32 hv){
    struct pcursor *cursor=&(db->pstack[hv].stack[db->pstack[hv].pnum-1]);
	struct rcd *rec=(struct rcd*)((char*)lf+lf->index[cursor->idxoff]);
	char buf[BLOCKSIZ];
	struct overflowpage *p;
	uint64 pgno;
	int siz;
	p=(struct overflowpage*)buf;
	DEBUG(printf("%s:append vsiz=%d idxoff=%d rec->vsiz=%d\n",__func__,vsiz,cursor->idxoff,rec->vsiz));
	if(rec->fap&V_OFLOW){
	    pgno=*((uint64*)rec->d);
		siz=rec->vsiz;
		rec->vsiz+=vsiz;
		
		while(siz>db->Pgsiz-OFLOWHEAD){
			seekread(db->Fd,buf,OFLOWHEAD,pgno);
			siz-=db->Pgsiz-OFLOWHEAD;
			pgno=p->nextpg;
		}
		if(BLOCKSIZ<db->Pgsiz) p=malloc(db->Pgsiz);
		seekread(db->Fd,(char*)p,db->Pgsiz,pgno);
		int wrsiz=vsiz<db->Pgsiz-siz-OFLOWHEAD?vsiz:db->Pgsiz-siz-OFLOWHEAD;
		char *ptr=val;
		memcpy((char*)p+siz,ptr,wrsiz);
		vsiz-=wrsiz;
		ptr+=wrsiz;
		p->siz=wrsiz+siz;
		while(vsiz>0){
			p->nextpg=getfpg(db);
			seekwrite(db->Fd,(char*)p,db->Pgsiz,pgno);
			wrsiz=vsiz>db->Pgsiz-OFLOWHEAD?db->Pgsiz-OFLOWHEAD:vsiz;
			memcpy((char*)p+OFLOWHEAD,ptr,wrsiz);
			ptr+=wrsiz;
			pgno=p->nextpg;
			p->siz=wrsiz;
			vsiz-=wrsiz;
		}
		p->nextpg=0;
		seekwrite(db->Fd,(char*)p,db->Pgsiz,pgno);
		putpage(db->cc,(char*)lf,F_DIRTY);
		if((char*)p!=buf) free(p);
	}else if(rec->fap&PK_OFLOW){
	    pgno=*((uint64*)rec->d);
		siz=rec->vsiz+rec->ksiz-(RECOVERSIZ-SIZEOFREC-sizeof(uint64));
		rec->vsiz+=vsiz;
		while(pgno!=0&&siz>db->Pgsiz-OFLOWHEAD){
			seekread(db->Fd,buf,OFLOWHEAD,pgno);
			siz-=db->Pgsiz-OFLOWHEAD;
			DEBUG(printf("%s:jump pgno=%d\n",__func__,pgno));
			pgno=p->nextpg;
		}
		if(BLOCKSIZ<db->Pgsiz) p=malloc(db->Pgsiz);
		seekread(db->Fd,(char*)p,db->Pgsiz,pgno);
		int wrsiz=vsiz<db->Pgsiz-siz-OFLOWHEAD?vsiz:db->Pgsiz-siz-OFLOWHEAD;
		char *ptr=val;
		memcpy((char*)p+OFLOWHEAD+siz,ptr,wrsiz);
		vsiz-=wrsiz;
		ptr+=wrsiz;
		p->siz=wrsiz+siz;
		while(vsiz>0){
			p->nextpg=getfpg(db);
			DEBUG(printf("%s:append rec pgno=%d bytes=%d\n",__func__,pgno,wrsiz));
			seekwrite(db->Fd,(char*)p,db->Pgsiz,pgno);
			wrsiz=vsiz>db->Pgsiz-OFLOWHEAD?db->Pgsiz-OFLOWHEAD:vsiz;
			memcpy((char*)p+OFLOWHEAD,ptr,wrsiz);
			ptr+=wrsiz;
			pgno=p->nextpg;
			p->siz=wrsiz;
			vsiz-=wrsiz;
		}
		p->nextpg=0;
		DEBUG(printf("%s:append rec pgno=%d bytes=%d\n",__func__,pgno,wrsiz));
		seekwrite(db->Fd,(char*)p,db->Pgsiz,pgno);
		putpage(db->cc,(char*)lf,F_DIRTY);
		if((char*)p!=buf) free(p);
	}
	else{
		char vbuf[BLOCKSIZ],*nval;
		int ilen=BLOCKSIZ,nvsiz;
		nval=vbuf;
		if(BLOCKSIZ<rec->vsiz+vsiz) nval=malloc(rec->vsiz+vsiz);
		getrecval(db,rec,nval,rec->vsiz);
		memcpy(nval+rec->vsiz,val,vsiz);
		nvsiz=rec->vsiz+vsiz;
		delrecord(db,lf,cursor->idxoff);
		rec=createrec(db,key,ksiz,nval,nvsiz,buf,&ilen);
		putrec(db,lf,rec,ilen,hv);
		if((char*)rec!=buf) free(rec);
		if(nval!=vbuf) free(nval);
	}
	return true;
}
int delindex(struct hdb *db,struct page *p,int16 ipos){
    char buf[BLOCKSIZ];
	struct page *bp=(struct page*)buf;
	if(db->Pgsiz>BLOCKSIZ) bp=malloc(db->Pgsiz);
	int k=0,i;
	uint16 inum=(p->lower-INDEXBEG)/sizeof(uint16);
	struct idx *tidx=(struct idx*)(p->index[ipos]+(char*)p);
	uint32 flag=tidx->fap&I_FLAGMASK;
	DEBUG(printf("%s:inum=%d ipos=%d\n",__func__,inum,ipos));
	if(flag&V_OFLOW){
	    uint64 pgno=*((uint64*)tidx->d);
		deloverflowpage(db,pgno,0,tidx->vsiz);
	}else if(flag&PK_OFLOW){
	    int pksiz=tidx->ksiz-(db->oversiz-SIZEOFIDX-sizeof(uint64));
	    deloverflowpage(db,*((uint64*)tidx->d),0,pksiz+tidx->vsiz);
	}
	bp->upper=db->Pgsiz;
	//if(ipos!=inum-1){
	for(i=0;i<inum;i++){
		if(i!=ipos){
		    struct idx *tidx=(struct idx*)((char*)p+p->index[i]);
			int ilen=padidx(db,tidx);
			bp->upper-=ilen;
			memcpy((char*)bp+bp->upper,tidx,ilen);
			DEBUG(printf("%s:cpy %d idx pgno=%d\n",__func__,i,tidx->pgno));
			bp->index[k++]=bp->upper;
		}
	}
	p->lower-=sizeof(int16);
	p->upper=bp->upper;
	memcpy(p->index,bp->index,p->lower-INDEXBEG);
	memcpy((char*)p+p->upper,(char*)bp+bp->upper,db->Pgsiz-bp->upper);
	//}
	//p->lower-=sizeof(uint16);
	//p->upper=upper-(char*)bp;
	if((char*)bp!=buf)free(bp);
	return true;
}
int delrecord(struct hdb *db,struct leaf *lf,int16 ipos){
    char buf[BLOCKSIZ];
	struct leaf *blf=(struct leaf*)buf;
	struct rcd *it;
	int rsiz,k=0,i;
	int inum=(lf->lower-LEAFIDXBEG)/sizeof(int16);
	struct rcd *rec=(struct rcd*)(lf->index[ipos]+(char*)lf);
	uint32 flag=rec->fap&I_FLAGMASK;
	if((flag&V_OFLOW)){
	    uint64 pgno=*((uint64*)rec->d);
		DEBUG(printf("%s:deloverflowpage flag=%x\n",__func__,flag));//getchar();
		deloverflowpage(db,pgno,0,rec->vsiz);
	}else if(flag&PK_OFLOW){
	    int pksiz=rec->ksiz-(RECOVERSIZ-SIZEOFREC-sizeof(uint64));
	    deloverflowpage(db,*((uint64*)rec->d),0,rec->vsiz+pksiz);
	}
	if(db->Pgsiz>BLOCKSIZ) blf=(struct leaf*)malloc(db->Pgsiz);
	DEBUG(printf("%s:inum=%d ipos=%d Pgsiz=%d\n",__func__,inum,ipos,db->Pgsiz));
	blf->upper=db->Pgsiz;
	for(i=0;i<inum;i++){
	    if(i!=ipos){
		    it=(struct rcd*)(lf->index[i]+(char*)lf);
			rsiz=sizeofrec(db,it);
			blf->upper-=rsiz;
			memcpy((char*)blf+blf->upper,it,rsiz);
			blf->index[k++]=blf->upper;
		}
	}
	lf->lower-=sizeof(int16);
	lf->upper=blf->upper;
	memcpy(lf->index,blf->index,lf->lower-LEAFIDXBEG);
	memcpy((char*)lf+lf->upper,(char*)blf+blf->upper,db->Pgsiz-blf->upper);
	if((char*)blf!=buf) free(blf);
	return true;
}
int hdbclose(struct hdb *db){
    int i;
	pthread_mutex_lock(&db->dblock);
	void *mapaddr=(char*)db->buckets-BUCKETBEG;
	int len=db->Bnum*sizeof(uint64)+BUCKETBEG;
	cachedel(db->cc);
	if(db->Nfpg>=((db->Maxpg/db->Pgsiz)>>1))
	    shrinkpage(db);
	for(i=0;i<db->Bnum;i++){
	    #ifdef __DB_DEBUG
		if(db->pstack[i].stack!=0)
	        printf("%s:bnum=%d i=%d db->pstack=%x pstack.stack=%x\n",__func__,db->Bnum,i,&db->pstack[i],db->pstack[i].stack);
		#endif
	    if(db->pstack[i].stack!=NULL) free(db->pstack[i].stack);
	    pthread_rwlock_destroy(db->block+i);
	}
	free(db->pstack);
	int fd=db->Fd;
	uint64 maxpg=db->Maxpg;
	if(msync(mapaddr,len,MS_SYNC)<0) fprintf(stderr,"msync:%s\n",strerror(errno));
	munmap(mapaddr,len);
	ftruncate(fd,maxpg);
	fsync(fd);
	DEBUG(printf("%s:Maxpg=%d\n",__func__,maxpg));
	close(fd);
	//fclose(db->fp);
	pthread_mutex_unlock(&db->dblock);
	pthread_mutex_destroy(&db->dblock);
	free(db->dname);
	free(db);
	return true;
}
int delrec(struct hdb *db,uint32 hv,struct leaf *lf,uint16 ipos){
	struct rcd *rec=(struct rcd*)((char*)lf+lf->index[ipos]);
	uint16 inum=(lf->lower-LEAFIDXBEG)/sizeof(uint16);
	struct page *p;
	struct idx *index;
	struct pgstack *pstack=db->pstack+hv;
	int i;
	uint64 pgno,npgno;
	STACKPOP(pstack,pgno,ipos);
	DEBUG(printf("%s:inum=%d ipos=%d stacknum=%d\n",__func__,inum,ipos,pstack->pnum));//getchar();
	if(inum>1){
	    delrecord(db,lf,ipos);
		putpage(db->cc,(char*)lf,F_DIRTY);
		return true;
	}
	if(ipos!=0){ printf("%s:odd!ipos=%d\n",__func__,ipos);exit(1);}
	if(rec->fap&V_OFLOW)
	    deloverflowpage(db,*((uint64*)rec->d),0,rec->vsiz);
	else if(rec->fap&PK_OFLOW){
	    int pksiz=rec->ksiz-(RECOVERSIZ-SIZEOFREC-sizeof(uint64));
	    deloverflowpage(db,*((uint64*)rec->d),0,rec->vsiz+pksiz);
	}
	freepg(db,PGNO(pgno));
	delpage(db->cc,(char*)lf);
	if(pstack->pnum==0){
	    printf("%s:root page is deleted\n",__func__);
	    db->buckets[hv]=0;
	    return true;
	}
	STACKPOP(pstack,pgno,ipos);
	p=(struct page*)getpage(db->cc,pgno,F_WRITE);
	inum=(p->lower-INDEXBEG)/sizeof(uint16);
	DEBUG(printf("%s:inum=%d ipos=%d\n",__func__,inum,ipos));
	while(inum==0&&pstack->pnum>=1){
	    freepg(db,pgno);
		delpage(db->cc,(char*)p);
		STACKPOP(pstack,pgno,ipos);
		p=(struct page*)getpage(db->cc,pgno,F_WRITE);
		inum=(p->lower-INDEXBEG)/sizeof(uint16);
		DEBUG(printf("%s:inum=%d ipos=%d\n",__func__,inum,ipos));
	}
	if(inum==0&&pstack->pnum==0){//root page
	    freepg(db,pgno);
		delpage(db->cc,(char*)p);
		db->buckets[hv]=0;
		DEBUG(printf("%s:root page deleted\n",__func__));
		return true;
	}
	else{//inum>0
		if(ipos==0){
		    index=(struct idx*)(p->index[0]+(char*)p);
			//p->left=index->pgno;
			npgno=index->pgno;
			DEBUG(printf("%s:left most pointer is to be deleted index->pgno=%d\n",__func__,index->pgno));//getchar();
			char *upper,buf[BLOCKSIZ],pbuf[BLOCKSIZ];
			int ilen=padidx(db,index)-sizeof(uint64),k=0;
			struct rcd *rec;
			struct page *bp;
			if(index->fap&I_DELETED){
			    delindex(db,p,0);
				//struct idx *iidx=(struct idx*)(p->index[0]+(char*)p);
				//uint64 pp=iidx->pgno;
			    //putpage(db->cc,(char*)p,F_DIRTY);
				DEBUG(printf("%s:the 0th index was deleted new 0 index pgno=%d\n",__func__,pgno));
				p->left=npgno;
				putpage(db->cc,(char*)p,F_DIRTY);
				return true;
			}
			if(ilen>BLOCKSIZ) rec=malloc(ilen);
			else rec=(struct rcd*)buf;
			IDXTOREC(index,ilen+sizeof(uint64),rec);
			if(db->Pgsiz>BLOCKSIZ) bp=malloc(db->Pgsiz);
			else bp=(struct page *)pbuf;
			upper=(char*)bp+db->Pgsiz;
			for(i=1;i<inum;i++){
		            struct idx *sidx=(struct idx*)((char*)p+p->index[i]);
			        int ilen=padidx(db,sidx);
			        upper-=ilen;
			        memcpy(upper,sidx,ilen);
			        bp->index[k++]=upper-(char*)bp;
	        }
	        memcpy(p->index,bp->index,k*sizeof(uint16));
	        memcpy((char*)p+(upper-(char*)bp),upper,(char*)bp+db->Pgsiz-upper);
	        p->lower-=sizeof(uint16);
	        p->upper=upper-(char*)bp;
			p->left=npgno;
			putpage(db->cc,(char*)p,F_DIRTY);
			STACKPUSH(pstack,pgno,0);
			//p=(struct page*)getpage(db->cc,PGNO(npgno),F_READ);
			while(!ISLEAF(npgno)){
			    DEBUG(printf("%s:locate pgno=%d\n",__func__,npgno));
			    p=(struct page*)getpage(db->cc,npgno,F_READ);
				STACKPUSH(pstack,npgno,0);
			    npgno=p->left;
				putpage(db->cc,(char*)p,0);
				//p=(struct page*)getpage(db->cc,PGNO(npgno),F_READ);
			}
			STACKPUSH(pstack,npgno,0);
			DEBUG(printf("%s:get leaf pgno=%d\n",__func__,npgno));
			lf=(struct leaf*)getpage(db->cc,PGNO(npgno),F_WRITE);
			DEBUG(printf("%s:leaf->inum=%d\n",__func__,(lf->lower-LEAFIDXBEG)/sizeof(uint16)));
			putrec(db,lf,rec,ilen,hv);
			return true;
		}else{
		    index=(struct idx*)(p->index[ipos-1]+(char*)p);
		    if(index->fap&I_DELETED){
			    DEBUG(printf("%s:%d idx was deleted before\n",__func__,ipos));
				#ifdef __DB_DEBUG
				int x;
				int in=(p->lower-INDEXBEG)/sizeof(int16);
				for(x=0;x<in;x++){
				    struct idx *iidx=(struct idx*)(p->index[x]+(char*)p);
					printf("%s:%dth idx->pgno=%d\n",__func__,x,iidx->pgno);
				}
				#endif
		        delindex(db,p,ipos-1);
				#ifdef __DB_DEBUG
				in=(p->lower-INDEXBEG)/sizeof(int16);
				if(in>0)
				for(x=0;x<in;x++){
				    struct idx *iidx=(struct idx*)(p->index[x]+(char*)p);
					printf("%s:%dth idx->pgno=%d\n",__func__,x,iidx->pgno);
				}else printf("%s:p->left=%d\n",__func__,p->left);
				#endif
			    putpage(db->cc,(char*)p,F_DIRTY);
			    return true;
		    }else{
			    DEBUG(printf("%s:%d index is to be inserted into %d index\n",__func__,ipos-1,ipos-2));
			    struct rcd *rec;
				char buf[BLOCKSIZ],pbuf[BLOCKSIZ],*upper;
			    int k=0,ilen=padidx(db,index)-sizeof(uint64);
				struct page *bp;
			    if(ilen>BLOCKSIZ) rec=malloc(ilen);
			    else rec=(struct rcd*)buf;
			    IDXTOREC(index,ilen+sizeof(uint64),rec);
				if(db->Pgsiz>BLOCKSIZ) bp=malloc(db->Pgsiz);
				else bp=(struct page*)pbuf;
				upper=(char*)bp+db->Pgsiz;
			    for(i=0;i<inum;i++){
		            if(i!=ipos-1){
		                struct idx *sidx=(struct idx*)((char*)p+p->index[i]);
			            int ilen=padidx(db,sidx);
			            upper-=ilen;
			            memcpy(upper,sidx,ilen);
			            bp->index[k++]=upper-(char*)bp;
		            }
	            }
	            memcpy(p->index,bp->index,k*sizeof(uint16));
	            memcpy((char*)p+(upper-(char*)bp),upper,(char*)bp+db->Pgsiz-upper);
	            p->lower-=sizeof(uint16);
	            p->upper=upper-(char*)bp;
				STACKPUSH(pstack,pgno,ipos-1);
				if(ipos-1==0)
					pgno=p->left;
				else
				    pgno=((struct idx*)(p->index[ipos-2]+(char*)p))->pgno;
				#ifdef __DB_DEBUG
				int in=(p->lower-INDEXBEG)/sizeof(int16);
				int x;
				for(x=0;x<in;x++){
				    struct idx *iidx=(struct idx*)(p->index[x]+(char*)p);
					printf("%s:%dth idx->pgno=%d\n",__func__,x,iidx->pgno);
				}
				#endif
				putpage(db->cc,(char*)p,F_DIRTY);
				while(!ISLEAF(pgno)){
				    p=(struct page*)getpage(db->cc,pgno,F_READ);
				    inum=(p->lower-INDEXBEG)/sizeof(uint16);
				    STACKPUSH(pstack,pgno,inum);
				    if(inum==0)
						pgno=p->left;
					else
					    pgno=((struct idx*)(p->index[inum-1]+(char*)p))->pgno;
					putpage(db->cc,(char*)p,0);
				}
				struct leaf *lf=(struct leaf*)getpage(db->cc,PGNO(pgno),F_WRITE);
				inum=(lf->lower-LEAFIDXBEG)/sizeof(uint16);
				DEBUG(printf("%s:going to insert into a leaf inum=%d\n",__func__,inum));
				STACKPUSH(pstack,pgno,inum);
				putrec(db,lf,rec,ilen,hv);
		    }
		}  
	}
	return true;
}
int hdberase(struct hdb *db,char *key,int ksiz){
    uint32 hv=hash1(db,key,ksiz);
	pthread_rwlock_wrlock(db->block+hv);
	uint64 pgno=db->buckets[hv];
	struct page *p;
	struct leaf *lf;
	struct idx *index;
	void *op;
	int hit=0;
	DEBUG(printf("%s:------------------hashval=%d\n",__func__,hv));
	if(pgno==0||(op=locate(db,key,ksiz,hv,&hit))==NULL||hit==0){
	    pthread_rwlock_unlock(db->block+hv);
		return false;
	}
	db->clock[hv]++;
	struct pgstack *pstack=&db->pstack[hv];
	struct pcursor *cursor=pstack->stack+pstack->pnum-1;
	DEBUG(printf("%s:locate stack pnum=%d\n",__func__,pstack->pnum);
	int x;
	for(x=0;x<pstack->pnum;x++){
	    printf("stack %d:pgno=%d idxoff=%d\n",x,pstack->stack[x].pgno,pstack->stack[x].idxoff);
	})
	int ipos=cursor->idxoff;
	if(ISLEAF(cursor->pgno)){
	    DEBUG(printf("%s:del from leaf ipos=%d stacknum=%d\n",__func__,ipos,pstack->pnum));//getchar();
	    lf=op;
	    delrec(db,hv,lf,ipos);
		putpage(db->cc,(char*)lf,F_DIRTY);
	}else{
	    p=(struct page*)op;
	    index=(struct idx*)((char*)p+p->index[cursor->idxoff]);
		if(index->fap&I_DELETED){
		    DEBUG(printf("%s:index is deleted\n",__func__));
			pthread_rwlock_unlock(db->block+hv);
			return true;
		}
		DEBUG(printf("%s:del from page ipos=%d\n",__func__,ipos));//getchar();
	    if((index->fap&PK_OFLOW)){
		    int pksiz=db->oversiz-sizeof(uint64)-SIZEOFIDX;
		    deloverflowpage(db,*((uint64*)index->d),pksiz,index->vsiz);
            index->vsiz=0;			
		}else if(index->fap&V_OFLOW){
		    deloverflowpage(db,*((uint64*)index->d),0,index->vsiz);
			index->vsiz=0;
		}
		index->fap|=I_DELETED;
		DEBUG(printf("%s:index->pgno=%d\n",__func__,index->pgno));
		putpage(db->cc,(char*)p,F_DIRTY);
	}
	pthread_rwlock_unlock(db->block+hv);
	pthread_mutex_lock(&db->dblock);
	db->Nrec--;
	pthread_mutex_unlock(&db->dblock);
	return true;
}
int hdbget(struct hdb *db,char *key,int ksiz,char *buf,int *bsiz){
    void *op;
	struct leaf* lf;
	struct page* pg;
	uint32 hv=hash1(db,key,ksiz);
	int hit=0,cpsiz,vsiz=0;
	DEBUG(printf("%s:key=%d hv=%d\n",__func__,xatoi(key,ksiz),hv));
	pthread_rwlock_rdlock(db->block+hv);
	if(db->buckets[hv]==0){
	    *bsiz=0;
		pthread_rwlock_unlock(db->block+hv);
	    return 0;
	}
	op=locate(db,key,ksiz,hv,&hit);
	if(hit){
		struct pgstack* pstack=db->pstack+hv;
		struct pcursor* cursor=pstack->stack+pstack->pnum-1;
		char vbuf[BLOCKSIZ];
		char *val=vbuf;
		char needfree=0;
		if(ISLEAF(cursor->pgno)){
		    lf=op;
			struct rcd* rec=(struct rcd*)((char*)lf+lf->index[cursor->idxoff]);
			uint32 flag=rec->fap&I_FLAGMASK;
			vsiz=rec->vsiz;
			if(flag==0) val=rec->d+rec->ksiz;
			else if(flag==V_OFLOW){
			    if(rec->vsiz>BLOCKSIZ){
				    val=malloc(rec->vsiz);
					needfree=1;
				}
			    getoverflowpage(db,*((uint64*)rec->d),0,val,rec->vsiz);
			}else if(flag==PK_OFLOW){
			    if(rec->vsiz>BLOCKSIZ){
				    val=malloc(rec->vsiz);
					needfree=1;
				}
				getoverflowpage(db,*((uint64*)rec->d),rec->ksiz-(RECOVERSIZ-SIZEOFREC-sizeof(uint64)),val,vsiz);
			}else{
			    printf("%s:bad rec fap=%d\n",__func__,rec->fap);
				exit(1);
			}
		}else{
		    pg=op;
			struct idx *index=(struct idx*)((char*)pg+pg->index[cursor->idxoff]);
			uint32 flag=index->fap&I_FLAGMASK;
			vsiz=index->vsiz;
			if(flag==0) val=index->d+index->ksiz;
			else if(flag==V_OFLOW){
			    if(index->vsiz>BLOCKSIZ){
				    val=malloc(index->vsiz);
				    needfree=1;
				}
				getoverflowpage(db,*((uint64*)index->d),0,val,index->vsiz);
			}else if(flag==PK_OFLOW){
			    if(index->vsiz>BLOCKSIZ){ 
				    val=malloc(index->vsiz);
					needfree=1;
				}
				getoverflowpage(db,*((uint64*)index->d),index->ksiz-(db->oversiz-SIZEOFIDX-sizeof(uint64)),val,vsiz);
			}else if(flag&I_DELETED){
			    DEBUG(printf("%s:this idx is deleted\n",__func__));
				vsiz=0;
			}
			else{
			    printf("%s:bad idx fap=%d\n",__func__,index->fap);
				exit(1);
			}
		}
		DEBUG(printf("%s:got vsiz=%d bufsiz=%d\n",__func__,vsiz,*bsiz));
		if(buf!=NULL){
		    cpsiz=*bsiz-1>vsiz?vsiz:*bsiz-1;
		    memcpy(buf,val,cpsiz);
		    buf[cpsiz]='\0';
		    *bsiz=cpsiz;
		}
		if(needfree) free(val);
		putpage(db->cc,op,0);
		pthread_rwlock_unlock(db->block+hv);
		return true;
	}
	putpage(db->cc,op,0);
	*bsiz=0;
	pthread_rwlock_unlock(db->block+hv);
	return vsiz;
}
int hdbget_ex(struct hdb *db,char *key,int ksiz,struct bstring **val){
	void *op;
	struct leaf* lf;
	struct page* pg;
	uint32 hv=hash1(db,key,ksiz);
	int hit=0;
	pthread_rwlock_rdlock(db->block+hv);
	if(val==NULL||db->buckets[hv]==0){
		pthread_rwlock_unlock(db->block+hv);
	    return 0;
	}
	op=locate(db,key,ksiz,hv,&hit);
	if(hit){
		struct pgstack* pstack=db->pstack+hv;
		struct pcursor* cursor=pstack->stack+pstack->pnum-1;
		if(ISLEAF(cursor->pgno)){
		    lf=op;
			struct rcd* rec=(struct rcd*)((char*)lf+lf->index[cursor->idxoff]);
			uint32 flag=rec->fap&I_FLAGMASK;
			if(*val==NULL) *val=strnew(rec->vsiz);
			else if((*val)->max<rec->vsiz) stradjust(*val,rec->vsiz);
			if(flag==0){
			    //val=strcreate(rec->d+rec->ksiz,rec->vsiz);
				memcpy((*val)->array,rec->d+rec->ksiz,rec->vsiz);
			}else if(flag==V_OFLOW){
			    //val=strnew(rec->vsiz);
			    getoverflowpage(db,*((uint64*)rec->d),0,(*val)->array,rec->vsiz);
				//val->siz=rec->vsiz;
			}else if(flag==PK_OFLOW){
			    //val=strnew(rec->vsiz);
				getoverflowpage(db,*((uint64*)rec->d),rec->ksiz-(RECOVERSIZ-SIZEOFREC-sizeof(uint64)),(*val)->array,rec->vsiz);
			    //val->siz=rec->vsiz;
			}else{
			    printf("%s:bad rec fap=%d\n",__func__,rec->fap);
				exit(1);
			}
			(*val)->siz=rec->vsiz;
		}else{
		    pg=op;
			struct idx *index=(struct idx*)((char*)pg+pg->index[cursor->idxoff]);
			uint32 flag=index->fap&I_FLAGMASK;
			//stradjust(val,index->vsiz);
			if(*val==NULL) *val=strnew(index->vsiz);
			else if((*val)->max<index->vsiz) stradjust(*val,index->vsiz);
			if(flag==0){
			    //val=strcreate(index->d+index->ksiz,index->vsiz);
				memcpy((*val)->array,index->d+index->ksiz,index->vsiz);
			}else if(flag==V_OFLOW){
			    //val=strnew(index->vsiz);
				getoverflowpage(db,*((uint64*)index->d),0,(*val)->array,index->vsiz);
				//val->siz=index->vsiz;
			}else if(flag==PK_OFLOW){
			    //val=strnew(index->vsiz);
				getoverflowpage(db,*((uint64*)index->d),index->ksiz-(db->oversiz-SIZEOFIDX-sizeof(uint64)),(*val)->array,index->vsiz);
				//val->siz=index->vsiz;
			}else if(flag&I_DELETED){
			     printf("%s:this idx is deleted\n",__func__);
			}else{
			    printf("%s:bad idx fap=%d\n",__func__,index->fap);
				exit(1);
			}
			(*val)->siz=index->vsiz;
		}
		putpage(db->cc,op,0);
		pthread_rwlock_unlock(db->block+hv);
		return 1;
	}
	putpage(db->cc,op,0);
	pthread_rwlock_unlock(db->block+hv);
	return 0;
}
struct bstring *hdbgetstr(struct hdb *db,struct bstring *key){
    void *op;
	struct leaf* lf;
	struct page* pg;
	struct bstring *val=NULL;
	uint32 hv=hash1(db,key->array,key->siz);
	int hit=0;
	pthread_rwlock_rdlock(db->block+hv);
	if(db->buckets[hv]==0){
		pthread_rwlock_unlock(db->block+hv);
	    return NULL;
	}
	op=locate(db,key->array,key->siz,hv,&hit);
	if(hit){
		struct pgstack* pstack=db->pstack+hv;
		struct pcursor* cursor=pstack->stack+pstack->pnum-1;
		if(ISLEAF(cursor->pgno)){
		    lf=op;
			struct rcd* rec=(struct rcd*)((char*)lf+lf->index[cursor->idxoff]);
			uint32 flag=rec->fap&I_FLAGMASK;
			//stradjust(val,rec->vsiz);
			if(flag==0){
			    val=strcreate(rec->d+rec->ksiz,rec->vsiz);
			}else if(flag==V_OFLOW){
			    val=strnew(rec->vsiz);
			    getoverflowpage(db,*((uint64*)rec->d),0,val->array,rec->vsiz);
				val->siz=rec->vsiz;
			}else if(flag==PK_OFLOW){
			    val=strnew(rec->vsiz);
				getoverflowpage(db,*((uint64*)rec->d),rec->ksiz-(RECOVERSIZ-SIZEOFREC-sizeof(uint64)),val->array,rec->vsiz);
			    val->siz=rec->vsiz;
			}else{
			    printf("%s:bad rec fap=%d\n",__func__,rec->fap);
				exit(1);
			}
		}else{
		    pg=op;
			struct idx *index=(struct idx*)((char*)pg+pg->index[cursor->idxoff]);
			uint32 flag=index->fap&I_FLAGMASK;
			//stradjust(val,index->vsiz);
			if(flag==0){
			    val=strcreate(index->d+index->ksiz,index->vsiz);
			}else if(flag==V_OFLOW){
			    val=strnew(index->vsiz);
				getoverflowpage(db,*((uint64*)index->d),0,val->array,index->vsiz);
				val->siz=index->vsiz;
			}else if(flag==PK_OFLOW){
			    val=strnew(index->vsiz);
				getoverflowpage(db,*((uint64*)index->d),index->ksiz-(db->oversiz-SIZEOFIDX-sizeof(uint64)),val->array,index->vsiz);
				val->siz=index->vsiz;
			}else if(flag&I_DELETED){
			     printf("%s:this idx is deleted\n",__func__);
			}else{
			    printf("%s:bad idx fap=%d\n",__func__,index->fap);
				exit(1);
			}
		}
		putpage(db->cc,op,0);
		pthread_rwlock_unlock(db->block+hv);
		return val;
	}
	putpage(db->cc,op,0);
	pthread_rwlock_unlock(db->block+hv);
	return NULL;
}
int nlocate(struct hdb *db,struct pgstack *pstack,struct bstring *lkey){
    char buf[BLOCKSIZ],*sbuf=NULL,*tbuf=NULL;
	struct page *p=NULL;
	struct leaf *lp=NULL;
	struct idx *index;
	struct rcd *rec;
	int left,right,mid,inum;
	int cmp,slen,len;
	uint32 hv=hash1(db,lkey->array,lkey->siz);
	uint64 pgno=db->buckets[hv];
	pstack->pnum=0;
	char *key=lkey->array;
	int ksiz=lkey->siz;
	while(!ISLEAF(pgno)){
	    p=(struct page*)getpage(db->cc,pgno,F_READ);
		inum=(p->lower-INDEXBEG)/sizeof(uint16);
		left=0;
		right=inum-1;
		DEBUG(printf("%s:internal page:%d inum=%d\n",__func__,pgno,inum));
		//getchar();
		while(left<=right){
		    mid=(left+right)>>1;
		    index=(struct idx*)(p->index[mid]+(char*)p);
			char *nkey;
		    int flag=index->fap&I_FLAGMASK&~I_DELETED;
		    if(flag==0||flag==V_OFLOW){
		        len=index->ksiz>ksiz?ksiz:index->ksiz;
			    nkey=flag==V_OFLOW?index->d+sizeof(uint64):index->d;
		        cmp=strncmp(key,nkey,len);
			    if(cmp==0) cmp=ksiz-index->ksiz;
		    }
		    else if(flag==PK_OFLOW){
		        len=db->oversiz-sizeof(uint64)-SIZEOFIDX>ksiz?ksiz:db->oversiz-sizeof(uint64)-SIZEOFIDX;
			    nkey=index->d+sizeof(uint64);
			    if((cmp=strncmp(key,nkey,len))==0){
			        if(ksiz>len){
					    if(index->ksiz-len>BLOCKSIZ){
						    if(sbuf==NULL){
						        sbuf=(char*)malloc(index->ksiz-len);
							    slen=index->ksiz-len;
						    }else if(slen<index->ksiz-len){
							    sbuf=(char*)realloc(sbuf,index->ksiz-len);
								slen=index->ksiz-len;
							}
							tbuf=sbuf;
						}else tbuf=buf;
				        getoverflowpage(db,*((uint64*)index->d),0,tbuf,index->ksiz-len);
					    int ln=len;
					    len=index->ksiz>ksiz?ksiz:index->ksiz;
					    len-=ln;
					    cmp=strncmp(key+ln,buf,len);
					    if(cmp==0) cmp=ksiz-index->ksiz;
				    }
				    else cmp=-1;
			    }
			}
			else{
			    printf("%s:error in locate key\n",__func__);
				exit(1);
			}
			if(cmp<0) right=mid-1;
            else if(cmp>0) left=mid+1;
            else{//find the same key
				STACKPUSH(pstack,pgno,mid);
			    if(tbuf!=buf&&tbuf!=NULL) free(tbuf);
				putpage(db->cc,(char*)p,0);
			    return true;
            }
		}
		STACKPUSH(pstack,pgno,left);
		if(left==0)
		    pgno=p->left;
		else
		    pgno=((struct idx*)(p->index[left-1]+(char*)p))->pgno;
		putpage(db->cc,(char*)p,0);
	}
	if(ISLEAF(pgno)){
		lp=(struct leaf*)getpage(db->cc,PGNO(pgno),F_READ);
		inum=(lp->lower-LEAFIDXBEG)/sizeof(uint16);
		left=0;
		right=inum-1;
		DEBUG(printf("%s:leaf page:%d inum=%d\n",__func__,pgno,inum));
		while(left<=right){
		    mid=(left+right)>>1;
		    rec=(struct rcd*)(lp->index[mid]+(char*)lp);
			DEBUG(printf("%s:mid=%d rec->fap=0x%x\n",__func__,mid,rec->fap));
			char *nkey;
		    int flag=rec->fap&I_FLAGMASK;
		    if(flag==0||flag==V_OFLOW){
		        len=rec->ksiz>ksiz?ksiz:rec->ksiz;
			    nkey=flag==V_OFLOW?rec->d+sizeof(uint64):rec->d;
		        cmp=strncmp(key,nkey,len);
			    if(cmp==0) cmp=ksiz-rec->ksiz;
		    }
		    else if(flag==PK_OFLOW){
		        len=RECOVERSIZ-SIZEOFREC-sizeof(uint64)>ksiz?ksiz:RECOVERSIZ-sizeof(uint64)-SIZEOFREC;
			    nkey=rec->d+sizeof(uint64);
			    if((cmp=strncmp(key,nkey,len))==0){
			        if(ksiz>len){
					    if(rec->ksiz-len>BLOCKSIZ){
						    if(sbuf==NULL){
						        sbuf=malloc(rec->ksiz-len);
							    slen=rec->ksiz-len;
							}else if(slen<rec->ksiz-len){
							    sbuf=realloc(sbuf,rec->ksiz-len);
								slen=rec->ksiz-len;
							}
							tbuf=sbuf;
						}else tbuf=buf;
				        getoverflowpage(db,*((uint64*)rec->d),0,tbuf,rec->ksiz-len);
					    int ln=len;
					    len=rec->ksiz>ksiz?ksiz:rec->ksiz;
					    len-=ln;
					    cmp=strncmp(key+ln,buf,len);
					    if(cmp==0) cmp=ksiz-index->ksiz;
				    }
				    else cmp=-1;
			    }
			}
			else{
			    printf("%s:error in locate key\n",__func__);
				exit(1);
			}
			if(cmp<0) right=mid-1;
            else if(cmp>0) left=mid+1;
            else{//find the same key
				STACKPUSH(pstack,pgno,mid);
			    if(tbuf!=buf&&tbuf!=NULL) free(tbuf);
			    putpage(db->cc,(char*)lp,0);
				return true;
            }
		}
		STACKPUSH(pstack,pgno,left);
	}else printf("%s:error!last page is not leaf\n",__func__);
	if(tbuf!=buf&&tbuf!=NULL) free(tbuf);
	putpage(db->cc,(char*)lp,0);
	return false;
}
struct iterator* iter_create(struct hdb* db){
    struct iterator* it=malloc(sizeof(struct iterator));
	int i=0;
	while(i<db->Bnum&&db->buckets[i]==0) i++;
    if(i==db->Bnum){
		free(it);
		return NULL;
	}
	it->db=db;
	it->key=strnew(0);
	it->pstack.stack=malloc(sizeof(struct pcursor)*DEF_LEVEL);
	it->pstack.pmax=DEF_LEVEL;
	it->pstack.pnum=0;
	it->bucket=-1;
	it->flag=IT_BEGIN|IT_FORWORD;
	return it;
}
int nextbucket(struct iterator *it){
    int bucket=it->bucket;
	struct hdb *db=it->db;
	struct page *p;
	uint64 pgno;
	while(++bucket<db->Bnum)
	    if(db->buckets[bucket]!=0) break;
	if(bucket==db->Bnum) return false;
	it->pstack.pnum=0;
	pgno=db->buckets[bucket];
	while(!ISLEAF(pgno)){
	    p=(struct page*)getpage(db->cc,pgno,F_READ);
		STACKPUSH(&it->pstack,pgno,0);
		pgno=p->left;
		putpage(db->cc,(char*)p,0);
	}
	STACKPUSH(&it->pstack,pgno,0);
	it->clock=db->clock[bucket];
	it->bucket=bucket;
	DEBUG(printf("%s:traverse bucket %d\n",__func__,bucket));
	return true;
}
int relocate(struct iterator *it){
	struct pgstack *pstack=&it->pstack;
	struct hdb *db=it->db;
	if(it->key->siz==0){
	    it->bucket=-1;
	    return nextbucket(it);
	}
	else
	    return nlocate(db,pstack,it->key);
}
int iter_next(struct iterator* it,struct bstring **key,struct bstring **val){
	struct hdb *db=it->db;
	struct pgstack *pstack=&it->pstack;
	int inum;
	uint64 pgno;
	struct leaf *lf=NULL,*tlf=NULL;
	struct page *p=NULL,*tp=NULL;
	struct rcd *rec;
	pthread_mutex_lock(&db->dblock);
	if(it->flag&IT_END){
	    pthread_mutex_unlock(&db->dblock);
	    return false;
	}
	if((it->flag&IT_BEGIN)){
		if(!nextbucket(it)){
		    pthread_mutex_unlock(&db->dblock);
		    return false;
		}
		it->flag&=~IT_BEGIN;
	}else{
	    //printf("%s:pstack->pnum=%d idxoff=%d\n",__func__,pstack->pnum,pstack->stack[pstack->pnum-1].idxoff);
	    if(it->clock!=db->clock[it->bucket]){
		    DEBUG(printf("%s:clock is different%d:%d\n",__func__,it->clock,db->clock[it->bucket]));
	        relocate(it);
		}
	    pgno=pstack->stack[pstack->pnum-1].pgno;
	    if(ISLEAF(pgno)){
	        lf=(struct leaf*)getpage(db->cc,PGNO(pgno),F_READ);
		    inum=(lf->lower-LEAFIDXBEG)/sizeof(int16);
			DEBUG(printf("%s:leaf inum=%d idxoff=%d pnum=%d\n",__func__,inum,pstack->stack[pstack->pnum-1].idxoff,pstack->pnum));
		    if(pstack->stack[pstack->pnum-1].idxoff>=inum){
			    putpage(db->cc,(char*)lf,0);
				if(pstack->pnum==1){
			        if(!nextbucket(it)){
					    pthread_mutex_unlock(&db->dblock);
					    return false;
					}
				}
			    else{
			        while(pstack->pnum>1){
			            pstack->pnum--;
					    pgno=pstack->stack[pstack->pnum-1].pgno;
					    p=(struct page*)getpage(db->cc,pgno,F_READ);
					    inum=(p->lower-INDEXBEG)/sizeof(int16);
					    if(pstack->stack[pstack->pnum-1].idxoff<inum){
						    tp=p;
							break;
						}
					    putpage(db->cc,(char*)p,0);
				    }
					DEBUG(printf("%s:pnum=%d\n",__func__,pstack->pnum));
				    if(pstack->pnum==1&&pstack->stack[pstack->pnum-1].idxoff>=inum){
				        if(!nextbucket(it)){
						    pthread_mutex_unlock(&db->dblock);
							return false;
						}
					}
					else{
					    p=(struct page*)getpage(db->cc,pstack->stack[pstack->pnum-1].pgno,F_READ);
						struct idx *tidx=(struct idx*)(p->index[pstack->stack[pstack->pnum-1].idxoff]+(char*)p);
						if(tidx->fap&I_DELETED){
						    pgno=tidx->pgno;
						    putpage(db->cc,(char*)p,0);
							while(!ISLEAF(pgno)){
							    p=(struct page*)getpage(db->cc,pgno,F_READ);
								STACKPUSH(pstack,pgno,0);
								pgno=p->left;
							}
							STACKPUSH(pstack,pgno,0);
						}else tp=p;
					}
		  	    }
		    }else tlf=lf;
	    }else{
	        p=(struct page*)getpage(db->cc,pgno,F_READ);
		    struct pcursor *cursor=pstack->stack+pstack->pnum-1;
		    struct idx *index=(struct idx*)(p->index[cursor->idxoff-1]+(char*)p);
		    pgno=index->pgno;
		    //pstack->stack[pstack->pnum-1].idxoff++;
		    while(!ISLEAF(pgno)){
		        putpage(db->cc,(char*)p,0);
			    p=(struct page*)getpage(db->cc,pgno,F_READ);
			    STACKPUSH(pstack,pgno,0);
			    pgno=p->left;
		    }
		    putpage(db->cc,(char*)p,0);
			STACKPUSH(pstack,pgno,0);
	    }
	}
	/*int x;
	printf("%s:bucket=%d\n",__func__,it->bucket);
	for(x=0;x<pstack->pnum;x++){
	    printf("%s:pgno=%d idxoff=%d\n",__func__,pstack->stack[x].pgno,pstack->stack[x].idxoff);
	}*/
	pgno=pstack->stack[pstack->pnum-1].pgno;
	if(ISLEAF(pgno)&&tlf==NULL){
	    tlf=(struct leaf*)getpage(db->cc,PGNO(pgno),F_READ);
	}else if(!ISLEAF(pgno)&&tp==NULL){
	    tp=(struct page*)getpage(db->cc,pgno,F_READ);
	}
	if(tlf!=NULL){
	    rec=(struct rcd*)(tlf->index[pstack->stack[pstack->pnum-1].idxoff]+(char*)tlf);
		uint32 flag=rec->fap&I_FLAGMASK;
		if(flag==0){
			*key=strcreate(rec->d,rec->ksiz);
			*val=strcreate(rec->d+rec->ksiz,rec->vsiz);
		}else if(flag==V_OFLOW){
			*key=strcreate(rec->d+sizeof(uint64),rec->ksiz);
			*val=strnew(rec->vsiz);
			getoverflowpage(db,*((uint64*)rec->d),0,(*val)->array,rec->vsiz);
			(*val)->siz=rec->vsiz;
		}else if(flag==PK_OFLOW){
			int pksiz=RECOVERSIZ-sizeof(uint64)-SIZEOFREC;
			char buf[BLOCKSIZ];
			char *tbuf=buf;
			*key=strnew(rec->ksiz);
			memcpy((*key)->array,rec->d+sizeof(uint64),pksiz);
			if(rec->ksiz-pksiz+rec->vsiz>BLOCKSIZ) tbuf=malloc(rec->ksiz-pksiz+rec->vsiz);
			getoverflowpage(db,*((uint64*)rec->d),0,tbuf,rec->ksiz+rec->vsiz-pksiz);
			memcpy((*key)->array+pksiz,tbuf,rec->ksiz-pksiz);
			(*key)->siz=rec->ksiz;
			*val=strcreate(tbuf+rec->ksiz-pksiz,rec->vsiz);
			if(tbuf!=buf) free(tbuf);
		}else{
		     printf("%s:bad leaf record fap=0x%x\n",__func__,rec->fap);
			 exit(1);
		}
		pstack->stack[pstack->pnum-1].idxoff++;
		putpage(db->cc,(char*)tlf,0);
	}else if(tp!=NULL){
	    struct idx* index=(struct idx*)(tp->index[pstack->stack[pstack->pnum-1].idxoff]+(char*)tp);
		uint32 flag=index->fap&I_FLAGMASK;
		if(flag==0){
		    //strassign(key,index->d,index->ksiz);
			//strassign(val,index->d+index->ksiz,index->vsiz);
			*key=strcreate(index->d,index->ksiz);
			*val=strcreate(index->d+index->ksiz,index->vsiz);
		}else if(flag==V_OFLOW){
		    //strassign(key,index->d+sizeof(uint64),index->ksiz);
			//stradjust(val,rec->vsiz);
			*key=strcreate(index->d+sizeof(uint64),index->ksiz);
			*val=strnew(rec->vsiz);
			getoverflowpage(db,*((uint64*)index->d),0,(*val)->array,index->vsiz);
			(*val)->siz=rec->vsiz;
		}else if(flag==PK_OFLOW){
		    int pksiz=db->oversiz-sizeof(uint64)-SIZEOFIDX;
			char buf[BLOCKSIZ];
			char *tbuf=buf;
			*key=strnew(index->ksiz);
			memcpy((*key)->array,index->d+sizeof(uint64),pksiz);
			if(index->ksiz-pksiz+index->vsiz>BLOCKSIZ) tbuf=malloc(index->ksiz-pksiz+index->vsiz);
			getoverflowpage(db,*((uint64*)index->d),0,tbuf,index->ksiz+index->vsiz-pksiz);
			memcpy((*key)->array+pksiz,tbuf,index->ksiz-pksiz);
			(*key)->siz=index->ksiz;
			*val=strcreate(tbuf+index->ksiz-pksiz,index->vsiz);
			if(tbuf!=buf) free(tbuf);
		}else{
		    printf("%s:bad fap=0x%x\n",__func__,index->fap);
			exit(1);
		}
		pstack->stack[pstack->pnum-1].idxoff++;
		putpage(db->cc,(char*)tp,0);
	}
	bstrdup(it->key,*key);
	pthread_mutex_unlock(&db->dblock);
	return true;
}
int iter_next_ex(struct iterator* it,struct bstring **key,struct bstring **val){
	struct hdb *db=it->db;
	struct pgstack *pstack=&it->pstack;
	int inum;
	uint64 pgno;
	struct leaf *lf=NULL,*tlf=NULL;
	struct page *p=NULL,*tp=NULL;
	struct rcd *rec;
	pthread_mutex_lock(&db->dblock);
	if(it->flag&IT_END){
	    pthread_mutex_unlock(&db->dblock);
	    return false;
	}
	if((it->flag&IT_BEGIN)){
		if(!nextbucket(it)){
		    pthread_mutex_unlock(&db->dblock);
		    return false;
		}
		it->flag&=~IT_BEGIN;
	}else{
	    //printf("%s:pstack->pnum=%d idxoff=%d\n",__func__,pstack->pnum,pstack->stack[pstack->pnum-1].idxoff);
	    if(it->clock!=db->clock[it->bucket]){
		    DEBUG(printf("%s:clock is different%d:%d\n",__func__,it->clock,db->clock[it->bucket]));
	        relocate(it);
		}
	    pgno=pstack->stack[pstack->pnum-1].pgno;
	    if(ISLEAF(pgno)){
	        lf=(struct leaf*)getpage(db->cc,PGNO(pgno),F_READ);
		    inum=(lf->lower-LEAFIDXBEG)/sizeof(int16);
			DEBUG(printf("%s:leaf inum=%d idxoff=%d pnum=%d\n",__func__,inum,pstack->stack[pstack->pnum-1].idxoff,pstack->pnum));
		    if(pstack->stack[pstack->pnum-1].idxoff>=inum){
			    putpage(db->cc,(char*)lf,0);
				if(pstack->pnum==1){
			        if(!nextbucket(it)){
					    pthread_mutex_unlock(&db->dblock);
					    return false;
					}
				}
			    else{
			        while(pstack->pnum>1){
			            pstack->pnum--;
					    pgno=pstack->stack[pstack->pnum-1].pgno;
					    p=(struct page*)getpage(db->cc,pgno,F_READ);
					    inum=(p->lower-INDEXBEG)/sizeof(int16);
					    if(pstack->stack[pstack->pnum-1].idxoff<inum){
						    tp=p;
							break;
						}
					    putpage(db->cc,(char*)p,0);
				    }
					DEBUG(printf("%s:pnum=%d\n",__func__,pstack->pnum));
				    if(pstack->pnum==1&&pstack->stack[pstack->pnum-1].idxoff>=inum){
				        if(!nextbucket(it)){
						    pthread_mutex_unlock(&db->dblock);
							return false;
						}
					}
					else{
					    p=(struct page*)getpage(db->cc,pstack->stack[pstack->pnum-1].pgno,F_READ);
						struct idx *tidx=(struct idx*)(p->index[pstack->stack[pstack->pnum-1].idxoff]+(char*)p);
						if(tidx->fap&I_DELETED){
						    pgno=tidx->pgno;
						    putpage(db->cc,(char*)p,0);
							while(!ISLEAF(pgno)){
							    p=(struct page*)getpage(db->cc,pgno,F_READ);
								STACKPUSH(pstack,pgno,0);
								pgno=p->left;
							}
							STACKPUSH(pstack,pgno,0);
						}else tp=p;
					}
		  	    }
		    }else tlf=lf;
	    }else{
	        p=(struct page*)getpage(db->cc,pgno,F_READ);
		    struct pcursor *cursor=pstack->stack+pstack->pnum-1;
		    struct idx *index=(struct idx*)(p->index[cursor->idxoff-1]+(char*)p);
		    pgno=index->pgno;
		    //pstack->stack[pstack->pnum-1].idxoff++;
		    while(!ISLEAF(pgno)){
		        putpage(db->cc,(char*)p,0);
			    p=(struct page*)getpage(db->cc,pgno,F_READ);
			    STACKPUSH(pstack,pgno,0);
			    pgno=p->left;
		    }
		    putpage(db->cc,(char*)p,0);
			STACKPUSH(pstack,pgno,0);
	    }
	}
	pgno=pstack->stack[pstack->pnum-1].pgno;
	if(ISLEAF(pgno)&&tlf==NULL){
	    tlf=(struct leaf*)getpage(db->cc,PGNO(pgno),F_READ);
	}else if(!ISLEAF(pgno)&&tp==NULL){
	    tp=(struct page*)getpage(db->cc,pgno,F_READ);
	}
	if(tlf!=NULL){
	    rec=(struct rcd*)(tlf->index[pstack->stack[pstack->pnum-1].idxoff]+(char*)tlf);
		uint32 flag=rec->fap&I_FLAGMASK;
		if(key!=NULL){
		    if(*key==NULL) *key=strnew(rec->ksiz);
			else if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
			(*key)->siz=rec->ksiz;
		}
		if(val!=NULL){
			if(*val==NULL) *val=strnew(rec->vsiz);
			else if((*val)->siz<rec->vsiz) stradjust(*val,rec->vsiz);
			(*val)->siz=rec->vsiz;
		}
		if(flag==0){
			if(key!=NULL) memcpy((*key)->array,rec->d,rec->ksiz);
			if(val!=NULL) memcpy((*val)->array,rec->d+rec->ksiz,rec->vsiz);
		}else if(flag==V_OFLOW){
			//*key=strcreate(rec->d+sizeof(uint64),rec->ksiz);
			//*val=strnew(rec->vsiz);
			if(key!=NULL) memcpy((*key)->array,rec->d,rec->ksiz);
			if(val!=NULL) getoverflowpage(db,*((uint64*)rec->d),0,(*val)->array,rec->vsiz);
			//(*val)->siz=rec->vsiz;
		}else if(flag==PK_OFLOW){
			int pksiz=RECOVERSIZ-sizeof(uint64)-SIZEOFREC;
			char buf[BLOCKSIZ];
			char *tbuf=buf;
			//*key=strnew(rec->ksiz);
			memcpy((*key)->array,rec->d+sizeof(uint64),pksiz);
			if(rec->ksiz-pksiz+rec->vsiz>BLOCKSIZ) tbuf=malloc(rec->ksiz-pksiz+rec->vsiz);
			getoverflowpage(db,*((uint64*)rec->d),0,tbuf,rec->ksiz+rec->vsiz-pksiz);
			memcpy((*key)->array+pksiz,tbuf,rec->ksiz-pksiz);
			//(*key)->siz=rec->ksiz;
			//*val=strcreate(tbuf+rec->ksiz-pksiz,rec->vsiz);
			memcpy((*val)->array,tbuf+rec->ksiz-pksiz,rec->vsiz);
			if(tbuf!=buf) free(tbuf);
		}else{
		     printf("%s:bad leaf record fap=0x%x\n",__func__,rec->fap);
			 exit(1);
		}
		pstack->stack[pstack->pnum-1].idxoff++;
		putpage(db->cc,(char*)tlf,0);
	}else if(tp!=NULL){
	    struct idx* index=(struct idx*)(tp->index[pstack->stack[pstack->pnum-1].idxoff]+(char*)tp);
		uint32 flag=index->fap&I_FLAGMASK;
		if(key!=NULL){
		    if(*key==NULL) *key=strnew(rec->ksiz);
			else if((*key)->siz<rec->ksiz) stradjust(*key,rec->ksiz);
			(*key)->siz=index->ksiz;
		}
		if(val!=NULL){
			if(*val==NULL) *val=strnew(rec->vsiz);
			else if((*val)->siz<rec->vsiz) stradjust(*val,rec->vsiz);
			(*val)->siz=index->vsiz;
		}
		if(flag==0){
		    //strassign(key,index->d,index->ksiz);
			//strassign(val,index->d+index->ksiz,index->vsiz);
			//*key=strcreate(index->d,index->ksiz);
			//*val=strcreate(index->d+index->ksiz,index->vsiz);
			if(key!=NULL) memcpy((*key)->array,index->d,index->ksiz);
			if(val!=NULL) memcpy((*val)->array,index->d+index->ksiz,index->vsiz);
		}else if(flag==V_OFLOW){
		    //strassign(key,index->d+sizeof(uint64),index->ksiz);
			//stradjust(val,rec->vsiz);
			if(key!=NULL) memcpy((*key)->array,index->d+sizeof(uint64),index->ksiz);
			//*key=strcreate(index->d+sizeof(uint64),index->ksiz);
			//*val=strnew(rec->vsiz);
			getoverflowpage(db,*((uint64*)index->d),0,(*val)->array,index->vsiz);
			//(*val)->siz=rec->vsiz;
		}else if(flag==PK_OFLOW){
		    int pksiz=db->oversiz-sizeof(uint64)-SIZEOFIDX;
			char buf[BLOCKSIZ];
			char *tbuf=buf;
			//*key=strnew(index->ksiz);
			memcpy((*key)->array,index->d+sizeof(uint64),pksiz);
			if(index->ksiz-pksiz+index->vsiz>BLOCKSIZ) tbuf=malloc(index->ksiz-pksiz+index->vsiz);
			getoverflowpage(db,*((uint64*)index->d),0,tbuf,index->ksiz+index->vsiz-pksiz);
			memcpy((*key)->array+pksiz,tbuf,index->ksiz-pksiz);
			//(*key)->siz=index->ksiz;
			//*val=strcreate(tbuf+index->ksiz-pksiz,index->vsiz);
			memcpy((*val)->array,tbuf+index->ksiz-pksiz,index->vsiz);
			if(tbuf!=buf) free(tbuf);
		}else{
		    printf("%s:bad fap=0x%x\n",__func__,index->fap);
			exit(1);
		}
		pstack->stack[pstack->pnum-1].idxoff++;
		putpage(db->cc,(char*)tp,0);
	}
	bstrdup(it->key,*key);
	pthread_mutex_unlock(&db->dblock);
	return true;
}
int iterdel(struct iterator *it){
    if(it->pstack.stack)
        free(it->pstack.stack);
	strdel(it->key);
	free(it);
	return true;
}
int shrinkpage(struct hdb *db){
    DEBUG(printf("%s:nfpg=%d Maxpg=%d\n",__func__,db->Nfpg,db->Maxpg));
	if(db->Nfpg>0){
	    char buf[BLOCKSIZ];
	    uint64 *parray=(uint64*)buf;
		int i=0;
		uint64 pgno;
		if(db->Nfpg*sizeof(uint64)>BLOCKSIZ)
		    parray=malloc(db->Nfpg*sizeof(uint64));
		if(db->Fstart>=0){
		    if(db->Fend<0){
			    memcpy(parray,db->fpa,db->Fnum*sizeof(uint64));
				i=db->Fnum;
			}
			else if(db->Fend>db->Fstart){
			    i=db->Fend-db->Fstart;
			    memcpy(parray,db->fpa+db->Fstart,i*sizeof(uint64));
			}
		    else if(db->Fend<db->Fstart){
			    i=db->Fnum-db->Fstart;
			    memcpy(parray,db->fpa+db->Fstart,i*sizeof(uint64));
			    memcpy(parray+i,db->fpa,db->Fend*sizeof(uint64));
			    i+=db->Fend;
		    }
		}
		if(db->Fread!=0){
		    struct freepage fp;
			pgno=db->Fread;
			DEBUG(printf("%s:fread=%d\n",__func__,db->Fread));
			while(pgno!=0&&i<db->Nfpg){
			    seekread(db->Fd,(char*)&fp,FREEHEAD,pgno);
				if(pgno<db->Maxpg)
				    parray[i++]=pgno;
				pgno=fp.nextpg;
				
			}
		}
		DEBUG(printf("%s:i=%d db->Nfpg=%d Maxpg=%d\n",__func__,i,db->Nfpg,db->Maxpg));
		if(db->Flast!=0){
		    struct freepage fp;
			pgno=db->Flast;
			DEBUG(printf("%s:flast pgno=%d\n",__func__,pgno));
			while(pgno!=0&&i<db->Nfpg){
			    seekread(db->Fd,(char*)&fp,FREEHEAD,pgno);
				if(pgno<db->Maxpg)
				    parray[i++]=pgno;
				pgno=fp.nextpg;
				
			}
		}
		DEBUG(printf("%s:new pgno=%d i=%d db->Nfpg=%d\n",__func__,pgno,i,db->Nfpg));
		sort64(parray,i);
		DEBUG(printf("%s:i=%d db->Nfpg=%d\n",__func__,i,db->Nfpg));
		while(parray[i-1]>db->Maxpg) i--;
		DEBUG(printf("%s:i=%d\n",__func__,i));
		while(i>0&&parray[i-1]==db->Maxpg-db->Pgsiz){
		    i--;
			db->Maxpg-=db->Pgsiz;
		}
		if(i==0){
		    db->Flast=db->Fread=0;
			db->Fstart=-1;
			db->Nfpg=0;
			return true;
		}
		DEBUG(printf("%s:i=%d\n",__func__,i));
		if(db->Nfpg>i){
		    i--;
		    db->Maxpg=parray[i];
		}
		int cp=i<db->Fnum?i:db->Fnum;
		memcpy(db->fpa,parray,cp*sizeof(uint64));
		db->Fstart=0;
		db->Fend=(cp==db->Fnum)?-1:cp;
		while(cp<i-1){
		    struct freepage fp={FREEPAGEMAGIC};
			fp.nextpg=parray[cp+1];
			DEBUG(printf("%s:cp=%d pgno=%d\n",__func__,cp,parray[cp]));
			seekwrite(db->Fd,(char*)&fp,sizeof(struct freepage),parray[cp]);
		    cp++;
		}
		if(cp==i-1){
		   struct freepage fp={FREEPAGEMAGIC};
		   fp.nextpg=0;
		   DEBUG(printf("%s:write last pgno=%d\n",__func__,parray[cp]));
		   seekwrite(db->Fd,(char*)&fp,sizeof(struct freepage),parray[cp]);
		   db->Flast=0;
		   db->Fread=parray[cp];
		}else{
		   db->Flast=0;
		   db->Fread=0;
		}
		db->Nfpg=i;
		DEBUG(printf("%s:nfpg=%d fstart=%d fend=%d flast=%d fread=%d\n",__func__,db->Nfpg,db->Fstart,db->Fend,db->Flast,db->Fread));
	}
	return true;
}
int htraverse(struct hdb *db,int root){
    if(root>db->Bnum||db->buckets[root]==0){
	    printf("bucket %d is null\n",root);
		return 0;
	}
	printf("%s:-------------bucket=%d\n",__func__,root);
	uint64 stack[100],off;
	int beg,end,inum,flag,siz=BLOCKSIZ,i;
	struct page* p;
	struct leaf *lf;
	struct rcd *rec;
	struct idx *index;
	char buf[BLOCKSIZ],*key;
	key=buf;
	stack[0]=db->buckets[root];
	beg=0,end=1;
	while(beg<end){
	    off=stack[beg];
		beg++;
		if(ISLEAF(off)){
		    lf=(struct leaf*)getpage(db->cc,PGNO(off),F_READ);
			inum=(lf->lower-LEAFIDXBEG)/sizeof(int16);
			printf("%s:leaf pgno=%d inum=%d\n",__func__,off,inum);
			for(i=0;i<inum;i++){
			    rec=(struct rcd*)((char*)lf+lf->index[i]);
				flag=rec->fap&I_FLAGMASK;
				if(rec->ksiz>siz){
				    if(key==buf)
				        key=malloc(rec->ksiz);
					else key=realloc(key,rec->ksiz);
					siz=rec->ksiz;
				}
				if(flag==0){
				    memcpy(key,(char*)rec+SIZEOFREC,rec->ksiz);
				}else if(flag==V_OFLOW){
				    memcpy(key,(char*)rec+SIZEOFREC+sizeof(uint64),rec->ksiz);
				}else if(flag==PK_OFLOW){
				    int pksiz=RECOVERSIZ-sizeof(uint64)-SIZEOFREC;
				    memcpy(key,(char*)rec+SIZEOFREC+sizeof(uint64),pksiz);
					getoverflowpage(db,*((uint64*)((char*)rec+SIZEOFREC)),0,key+pksiz,rec->ksiz-pksiz);
				}else{
				    printf("%s:bad rec fap=%x\n",__func__,rec->fap);
					exit(1);
				}
				printf("%s:%d key=%d\n",__func__,i,xatoi(key,rec->ksiz));
			}
			putpage(db->cc,(char*)lf,0);
		}else{
		    p=(struct page*)getpage(db->cc,off,F_READ);
			stack[end++]=p->left;
			inum=(p->lower-INDEXBEG)/sizeof(int16);
			printf("%s:pgno=%d inum=%d left=%d\n",__func__,off,inum,p->left);
			for(i=0;i<inum;i++){
			    index=(struct idx*)((char*)p+p->index[i]);
				flag=index->fap&I_FLAGMASK;
				if(index->ksiz>siz){
				    if(key==buf)
				        key=malloc(rec->ksiz);
					else key=realloc(key,index->ksiz);
					siz=index->ksiz;
				}
				if(flag==0){
				    memcpy(key,(char*)index+SIZEOFIDX,index->ksiz);
				}else if(flag==V_OFLOW){
				    memcpy(key,(char*)index+SIZEOFIDX+sizeof(uint64),index->ksiz);
				}else if(flag==PK_OFLOW){
				    int pksiz=db->oversiz-sizeof(uint64*)-SIZEOFIDX;
					memcpy(key,(char*)index+SIZEOFIDX+sizeof(uint64),pksiz);
					getoverflowpage(db,*((uint64*)((char*)index+SIZEOFIDX)),0,key+pksiz,index->ksiz-pksiz);
				}else{
				    printf("%s:bad index:fap=%x\n",__func__,index->fap);
					exit(1);
				}
				printf("%s:%d key=%d pgno=%d\n",__func__,i,xatoi(key,index->ksiz),index->pgno);
				stack[end++]=index->pgno;
			}
			putpage(db->cc,(char*)p,0);
		}
	}
	return true;
}
void htraver(struct hdb *db){
    int i;
	for(i=0;i<db->Bnum;i++){
	   htraverse(db,i);
	}
}
int hdbdelete(struct hdb *db){
    int dlen=strlen(db->dname);
	char *buf=malloc(dlen+1);
	if(buf==NULL) return false;
	memcpy(buf,db->dname,dlen);
	buf[dlen]='\0';
	hdbclose(db);
	unlink(buf);
	return true;
}
int hdbtrans_beg(struct hdb *db){
    pthread_rwlock_wrlock(&db->cc->lock);
    cachesyn(db->cc);
	if(db->newpgno==NULL){
	    STACK_NEW(db->newpgno,uint64);
	}
	if(db->delpgno==NULL){
	    STACK_NEW(db->delpgno,uint64);
	}
	db->flag|=HDB_IN_TRANSACTION;
	TRANSACTION_SET(db->cc);
	return true;
}
int hdbtrans_commit(struct hdb *db){
    cachesyn(db->cc);
	TRANSACTION_UNSET(db->cc);
	db->flag&=~HDB_IN_TRANSACTION;
	uint64 pgno;
	STACK_POP(db->delpgno,uint64,pgno);
	while(pgno!=0){
	    freepg(db,pgno);
		STACK_POP(db->delpgno,uint64,pgno);
	}
	STACK_CLEAR(db->newpgno);
	STACK_CLEAR(db->delpgno);
	pthread_rwlock_unlock(&db->cc->lock);
	return true;
}
int hdbtrans_abort(struct hdb *db){
    cacheabort(db->cc);
	TRANSACTION_UNSET(db->cc);
	db->flag&=~HDB_IN_TRANSACTION;
	uint64 pgno;
	STACK_POP(db->newpgno,uint64,pgno);
	while(pgno!=0){
	    freepg(db,pgno);
		int i;//FIXME...
		for(i=0;i<db->Bnum;i++){//bad implement..
		    if(PGNO(db->buckets[i])==pgno) db->buckets[i]=0;
		}
		STACK_POP(db->newpgno,uint64,pgno);
	}
	STACK_CLEAR(db->newpgno);
	STACK_CLEAR(db->delpgno);
	pthread_rwlock_unlock(&db->cc->lock);
	return true;
}
int hdbtrans_cancel(struct hdb *db){
    TRANSACTION_UNSET(db->cc);
	db->flag&=~HDB_IN_TRANSACTION;
    pthread_rwlock_unlock(&db->cc->lock);
	return true;
}
